8 Oracle Streams Advanced Queuing

Overview of Oracle Streams Advanced Queuing

Oracle Streams is a new information sharing feature that provides replication, message queuing, data warehouse loading, and event notification. It is also the foundation behind Oracle Streams Advanced Queuing (AQ).

Advanced Queuing is the integrated message queuing feature that exposes message queuing capabilities of Oracle Streams. AQ enables applications to:

• Perform message queuing operations similar to SQL operations from the Oracle database

• Communicate asynchronously through messages in AQ queues

• Integrate with database for unprecedented levels of operational simplicity, reliability, and security to message queuing

• Audit and track messages

• Supports both synchronous and asynchronous modes of communication

The advantages of using AQ in OCCI applications include:

• Create applications that communicate with each other in a consistent, reliable, secure, and autonomous manner

• Store messages in database tables, bringing the reliability and recoverability of the database to your messaging infrastructure

• Retain messages in the database automatically for auditing and business intelligence

• Create applications that leverage messaging without having to deal with a different security, data type, or operational mode

• Leverage transactional characteristics of the database

Since traditional messaging solutions have single subscriber queues, a queue must be created for each pair of applications that communicate with each other. The publish/subscribe protocol of the AQ makes it easy to add additional applications (subscribers) to a conversation between multiple applications.

AQ Implementation in OCCI

OCCI AQ is a set of interfaces that allows messaging clients to access the Advanced Queuing feature of Oracle for enterprise messaging applications. Currently, OCCI AQ supports only the operational interfaces and not the administrative interface, but administrative operations can be accessed through embedded PL/SQL calls.

The AQ feature can be used in conjunction with other interfaces available through OCCI for sending, receiving, publishing, and subscribing in a message-enabled database. Synchronous and asynchronous message consumption is available based on a message selection rule.

Enqueuing refers to sending a message to a queue and dequeuing refers to receiving one. A client application can create a message, set the desired properties on it and enqueue it by storing the message in the queue, a table in the database. When dequeuing a message, an application can either dequeue it synchronously by calling receive methods on the queue, or asynchronously by waiting for a notification from the database.

The AQ feature is implemented through the following abstractions:

• Message

• Agent

• Producer

• Consumer

• Listener

Agent agt(env, "Billing_app", "billqueue", 0);

Creating Messages

As mentioned previously, a Message is a basic unit of information that contains both the properties of the message and its content, or payload. Each message is enqueued by the Producer and dequeued by the Consumer objects.

AnyData any(conn);
any.setFromString("item1");
Message mes(env);
mes.setAnyData(any);

TypeCode tc = any.getType();

// Assuming type Employee ( name varchar2(25),
//                          deptid number(10),
//                          manager varchar2(25) )
Employee *emp = new Employee();
emp.setName("Scott");
emp.setDeptid(10);
emp.setManager("James");
Message mes(env);
mes.setObject(emp);

mes.setCorrelationId("enq_corr_di");

mes.setSenderId(agt);

mes.setDelay(10);
mes.setExpirationTime(60);

vector<Agent> agt_list;
for (i=0; i<num_recipients; i++)
   agt_list.push_back(Agent(name, address, protocol));
mes.setRecipientList(agt_list);

mes.setPriority(3);

Enqueuing Messages

Messages are enqueued by the Producer. The Producer Class is also used to specify enqueue options. A Producer object can be created on a valid connection where enqueuing will be performed, as illustrated in Example 8-10.

The transactional behavior of the enqueue operation can be defined based on application requirements. The application can make the effect of the enqueue operation visible externally either immediately after it is completed, as in Example 8-10, or only after the enclosing transaction has been committed.

To enqueue the message, use the send() method, as demonstrated in Example 8-10. A client may retain the Message object after it is sent, modify it, and send it again.

Producer prod(conn);
...
prod.setVisibility(Producer::ENQ_IMMEDIATE);
...
prod.send(mes, queueName);

Dequeuing Messages

Messages delivered to a queue are dequeued by the Consumer. The Consumer Class is also used to specify dequeue options. A Consumer object can be created on a valid connection to the database where a queue exists, as demonstrated in Example 8-11.

In applications that support multiple consumers in the same queue, the name of the consumer has to be specified as a registered subscriber to the queue, as shown in Example 8-11.

To dequeue the message, use the receive() method, as demonstrated in Example 8-11. The typeName and schemaName parameters of the receive() method specify the type of payload and the schema of the payload type.

Consumer cons(conn);
...
cons.setConsumerName("BillApp");
cons.setQueueName(queueName);
...
Message mes = cons.receive(Message::OBJECT, "BILL_TYPE", "BILL_PROCESSOR");

When the queue payload type is either RAW or AnyData, schemaName and typeName are optional, but you must specify these parameters explicitly when working with user-defined payloads. This is illustrated in Example 8-12.

//receiving a RAW message
Message mes = cons.receive(Message::RAW);
...
//receiving an ANYDATA message
Message mes = cons.receive(Message::ANYDATA);
...

cons.setCorrelationId(corrId);
...
cons.setDequeueMode(deqMode);
...
cons.setPositionOfMessage(Consumer::DEQ_NEXT_TRANSACTION);

Listening for Messages

The Listener listens for messages on queues on behalf of its registered clients. The Listener Class implements the listen() method, which is a blocking call that returns once a queue has a message for one of the registered agents, or throws an error when the time out period expires. Example 8-14 illustrates the listening protocol.

Listener listener(conn);

vector<Agent> agtList;
for( int i=0; i<num_agents; i++)
   agtList.push_back( Agent( name, address, protocol);

listener.setAgentList(agtList);
listener.setTimeOutForListen(10);

Agent agt(env);

try{
   agt = listener.listen();
}
catch{
   cout<<e.getMessage()<<endl;
}

Registering for Notification

The Subscription Class implements the publish-subscribe notification feature. It allows an OCCI AQ application to receive client notifications directly, register an e-mail address to which notifications can be sent, register an HTTP URL to which notifications can be posted, or register a PL/SQL procedure to be invoked on a notification. Registered clients are notified asynchronously when events are triggered or on an explicit AQ enqueue. Clients do not need to be connected to a database.

An OCCI application can do all of the following:

• Register interest in notification in the AQ namespace, and be notified when an enqueue occurs.

• Register interest in subscriptions to database events, and receive notifications when these events are triggered.

• Manage registrations, such as disable registrations temporarily, or dropping registrations entirely.

• Post (or send) notifications to registered clients.

ALTER SYSTEM REFRESH LDAP_REGISTRATION

typedef unsigned int (*callbackfn) (Subscription &sub, NotifyResult *nr);

Message Format Transformation

Applications often use data in different formats, and this requires a type transformation. A transformation is implemented as a SQL function that takes the source data type as input and returns an object of the target data type.Transformations can be applied when message are enqueued, dequeued, or when they are propagated to a remote subscriber.