8 Oracle Streams Advanced Queuing Administrative Interface

Creating a Queue Table

DBMS_AQADM.CREATE_QUEUE_TABLE(
   queue_table          IN      VARCHAR2,
   queue_payload_type   IN      VARCHAR2,
   [storage_clause      IN      VARCHAR2        DEFAULT NULL,]
   sort_list            IN      VARCHAR2        DEFAULT NULL,
   multiple_consumers   IN      BOOLEAN         DEFAULT FALSE,
   message_grouping     IN      BINARY_INTEGER  DEFAULT NONE,
   comment              IN      VARCHAR2        DEFAULT NULL,
   primary_instance     IN      BINARY_INTEGER  DEFAULT 0, 
   secondary_instance   IN      BINARY_INTEGER  DEFAULT 0,
   compatible           IN      VARCHAR2        DEFAULT NULL,
   secure               IN      BOOLEAN         DEFAULT FALSE);

This procedure creates a queue table for messages of a predefined type. It has the following required and optional parameters:

Payload Type

To specify the payload type as an object type, you must define the object type.

CLOB, BLOB, and BFILE objects are valid in an Oracle Streams Advanced Queuing message. You can propagate these object types using Oracle Streams Advanced Queuing propagation with Oracle software since Oracle8i release 8.1.x. To enqueue an object type that has a LOB, you must first set the LOB_attribute to EMPTY_BLOB() and perform the enqueue. You can then select the LOB locator that was generated from the queue table's view and use the standard LOB operations.

Storage Clause

The storage_clause argument can take any text that can be used in a standard CREATE TABLE storage_clause argument.

Once you pick the tablespace, any index-organized table (IOT) or index created for that queue table goes to the specified tablespace. You do not currently have a choice to split them between different tablespaces.

ALTER TABLE AQ$_queue_table_I COALESCE;

Sort Key

The sort_list parameter determines the order in which messages are dequeued. You cannot change the message sort order after you have created the queue table. Your choices are:

• ENQ_TIME

• ENQ_TIME,PRIORITY

• PRIORITY

• PRIORITY,ENQ_TIME

• PRIORITY,COMMIT_TIME

• COMMIT_TIME

The COMMIT_TIME choice is a new feature in Oracle Streams Advanced Queuing 10g Release 2 (10.2). If it is specified, then any queue that uses the queue table is a commit-time queue, and Oracle Streams Advanced Queuing computes an approximate CSCN for each enqueued message when its transaction commits.

If you specify COMMIT_TIME as the sort key, then you must also specify the following:

• multiple_consumers = TRUE

• message_grouping = TRANSACTIONAL

• compatible = 8.1 or higher

Commit-time ordering is useful when transactions are interdependent or when browsing the messages in a queue must yield consistent results.

Other Tables and Views

The following objects are created at table creation time:

• AQ$_queue_table_name, a read-only view which is used by Oracle Streams Advanced Queuing applications for querying queue data

• AQ$_queue_table_name_E, the default exception queue associated with the queue table

• AQ$_queue_table_name_I, an index or an index-organized table (IOT) in the case of multiple consumer queues for dequeue operations

• AQ$_queue_table_name_T, an index for the queue monitor operations

The following objects are created only for 8.1-compatible multiconsumer queue tables:

• AQ$_queue_table_name_S, a table for storing information about subscribers

• AQ$_queue_table_name_H, an index organized table (IOT) for storing dequeue history data

• AQ$_queue_table_name_L, dequeue log table, used for storing message identifiers of committed dequeue operations on the queue

If you do not specify a schema, then you default to the user's schema.

If GLOBAL_TOPIC_ENABLED = TRUE when a queue table is created, then a corresponding Lightweight Directory Access Protocol (LDAP) entry is also created.

If the queue type is ANYDATA, then a buffered queue and two additional objects are created. The buffered queue stores logical change records created by a capture process. The logical change records are staged in a memory buffer associated with the queue; they are not ordinarily written to disk.

If they have been staged in the buffer for a period of time without being dequeued, or if there is not enough space in memory to hold all of the captured events, then they are spilled to:

• AQ$_queue_table_name_P, a table for storing the captured events that spill from memory

• AQ$_queue_table_name_D, a table for storing information about the propagations and apply processes that are eligible for processing each event

Examples

The following examples assume you are in a SQL*Plus testing environment. In Example 8-1, you create users in preparation for the other examples in this chapter. For this example, you must connect as a user with administrative privileges. For most of the other examples in this chapter, you can connect as user test_adm. A few examples must be run as test with EXECUTE privileges on DBMS_AQADM.

CREATE USER test_adm IDENTIFIED BY test_adm DEFAULT TABLESPACE example;
GRANT DBA, CREATE ANY TYPE TO test_adm; 
GRANT EXECUTE ON DBMS_AQADM TO test_adm;
GRANT aq_administrator_role TO test_adm;
BEGIN
   DBMS_AQADM.GRANT_SYSTEM_PRIVILEGE(
      privilege          =>    'MANAGE_ANY', 
      grantee            =>    'test_adm', 
      admin_option       =>     FALSE);
END;
/
CREATE USER test IDENTIFIED BY test;
GRANT DBA TO test; 
GRANT EXECUTE ON dbms_aq TO test;

CREATE TYPE test.message_typ AS object(
   sender_id              NUMBER,
   subject                VARCHAR2(30),
   text                   VARCHAR2(1000));
/
CREATE TYPE test.msg_table AS TABLE OF test.message_typ;
/
CREATE TYPE test.order_typ AS object(
   custno                 NUMBER,
   item                   VARCHAR2(30),
   description            VARCHAR2(1000));
/
CREATE TYPE test.lob_typ AS object(
   id                     NUMBER, 
   subject                VARCHAR2(100), 
   data                   BLOB, 
   trailer                NUMBER);
/ 

BEGIN
   DBMS_AQADM.CREATE_QUEUE_TABLE(
      queue_table            => 'test.obj_qtab',
      queue_payload_type     => 'test.message_typ');
END;
/

BEGIN
   DBMS_AQADM.CREATE_QUEUE_TABLE( 
      queue_table            => 'test.raw_qtab', 
      queue_payload_type     => 'RAW'); 
END;
/

BEGIN
   DBMS_AQADM.CREATE_QUEUE_TABLE(
      queue_table            => 'test.lob_qtab',
      queue_payload_type     => 'test.lob_typ');
END;
/

BEGIN
  DBMS_AQADM.CREATE_QUEUE_TABLE(
     queue_table         => 'test.xml_qtab', 
     queue_payload_type  => 'SYS.XMLType',
     multiple_consumers  => TRUE, 
     compatible          => '8.1',
     comment             => 'Overseas Shipping multiconsumer orders queue table');
END;
/

BEGIN
   DBMS_AQADM.CREATE_QUEUE_TABLE( 
      queue_table          => 'test.group_qtab',
      queue_payload_type   => 'test.message_typ',
      message_grouping     => DBMS_AQADM.TRANSACTIONAL);
END;
/

BEGIN
   DBMS_AQADM.CREATE_QUEUE_TABLE(
      queue_table            => 'test.priority_qtab', 
      queue_payload_type     => 'test.order_typ',
      sort_list              => 'PRIORITY,ENQ_TIME', 
      multiple_consumers     => TRUE);
   DBMS_AQADM.CREATE_QUEUE_TABLE(
      queue_table            => 'test.multiconsumer_qtab',
      queue_payload_type     => 'test.message_typ',
      sort_list              => 'PRIORITY,ENQ_TIME', 
      multiple_consumers     => TRUE);
END;
/

BEGIN
   DBMS_AQADM.CREATE_QUEUE_TABLE(
      queue_table            => 'test.commit_time_qtab',
      queue_payload_type     => 'test.message_typ',
      sort_list              => 'COMMIT_TIME', 
      multiple_consumers     => TRUE, 
      message_grouping       => DBMS_AQADM.TRANSACTIONAL,
      compatible             => '10.0');
END;
/

BEGIN
   DBMS_AQADM.CREATE_QUEUE_TABLE(
      queue_table            => 'test.multiconsumer_81_qtab',
      queue_payload_type     => 'test.message_typ',
      multiple_consumers     =>  TRUE,
      compatible             => '8.1'); 
END;
/

BEGIN
   DBMS_AQADM.CREATE_QUEUE_TABLE( 
      queue_table        => 'test.example_qtab',
      queue_payload_type => 'test.message_typ',
      storage_clause     => 'tablespace example');
END;
/

BEGIN 
   DBMS_AQADM.CREATE_QUEUE_TABLE( 
      queue_table            => 'test.freelist_qtab', 
      queue_payload_type     => 'RAW', 
      storage_clause         => 'STORAGE (FREELISTS 4 FREELIST GROUPS 2)',
      compatible             => '8.1');
END;
/

Altering a Queue Table

DBMS_AQADM.ALTER_QUEUE_TABLE (
   queue_table          IN   VARCHAR2, 
   comment              IN   VARCHAR2       DEFAULT NULL,
   primary_instance     IN   BINARY_INTEGER DEFAULT NULL, 
   secondary_instance   IN   BINARY_INTEGER DEFAULT NULL);

This procedure alters the existing properties of a queue table.

If GLOBAL_TOPIC_ENABLED = TRUE when a queue table is modified, then a corresponding LDAP entry is also altered.

BEGIN
   DBMS_AQADM.ALTER_QUEUE_TABLE( 
      queue_table          => 'test.obj_qtab', 
      primary_instance     => 3, 
      secondary_instance   => 2);
END;
/

BEGIN
   DBMS_AQADM.ALTER_QUEUE_TABLE( 
      queue_table          => 'test.obj_qtab', 
      comment              => 'revised usage for queue table');
END;
/

Dropping a Queue Table

DBMS_AQADM.DROP_QUEUE_TABLE(
   queue_table       IN    VARCHAR2,
   force             IN    BOOLEAN DEFAULT FALSE,

This procedure drops an existing queue table. You must stop and drop all the queues in a queue table before the queue table can be dropped. You must do this explicitly if force is set to FALSE. If force is set to TRUE, then all queues in the queue table and their associated propagation schedules are dropped automatically.

If GLOBAL_TOPIC_ENABLED = TRUE when a queue table is dropped, then a corresponding LDAP entry is also dropped.

BEGIN
   DBMS_AQADM.DROP_QUEUE_TABLE(
      queue_table        => 'test.obj_qtab');
END;
/

BEGIN
   DBMS_AQADM.DROP_QUEUE_TABLE( 
      queue_table        => 'test.raw_qtab', 
      force              => TRUE); 
END;
/

Purging a Queue Table

DBMS_AQADM.PURGE_QUEUE_TABLE(
   queue_table        IN   VARCHAR2,
   purge_condition    IN   VARCHAR2,
   purge_options      IN   aq$_purge_options_t);

This procedure purges messages from a queue table. It has the following parameters:

A trace file is generated in the udump destination when you run this procedure. It details what the procedure is doing. The procedure commits after it has processed all the messages.

DECLARE
po dbms_aqadm.aq$_purge_options_t;
BEGIN
   po.block := FALSE;
   DBMS_AQADM.PURGE_QUEUE_TABLE(
     queue_table     => 'test.obj_qtab',
     purge_condition => NULL,
     purge_options   => po);
END;
/

DECLARE
po dbms_aqadm.aq$_purge_options_t;
BEGIN
   po.block := TRUE;
   DBMS_AQADM.PURGE_QUEUE_TABLE(
     queue_table     => 'test.obj_qtab',
     purge_condition => 'qtview.queue = ''TEST.OBJ_QUEUE''',
     purge_options   => po);
END;
/

DECLARE
po dbms_aqadm.aq$_purge_options_t;
BEGIN
   po.block := TRUE;
   DBMS_AQADM.PURGE_QUEUE_TABLE(
     queue_table     => 'test.obj_qtab',
     purge_condition => 'qtview.queue = ''TEST.OBJ_QUEUE'' 
                         and qtview.msg_state = ''PROCESSED''',
     purge_options   => po);
END;
/

DECLARE
po dbms_aqadm.aq$_purge_options_t;
BEGIN
   po.block := TRUE;
   DBMS_AQADM.PURGE_QUEUE_TABLE(
     queue_table     => 'test.multiconsumer_81_qtab',
     purge_condition => 'qtview.queue = ''TEST.MULTICONSUMER_81_QUEUE'' 
                         and qtview.consumer_name = ''PAYROLL_APP''',
     purge_options   => po);
END;
/

DECLARE
po dbms_aqadm.aq$_purge_options_t;
BEGIN
   po.block := TRUE;
   DBMS_AQADM.PURGE_QUEUE_TABLE(
     queue_table     => 'test.multiconsumer_81_qtab',
     purge_condition => 'qtview.sender_name = ''TEST.OBJ_QUEUE''',
     purge_options   => po);
END;
/

Migrating a Queue Table

DBMS_AQADM.MIGRATE_QUEUE_TABLE(
   queue_table   IN   VARCHAR2,
   compatible    IN   VARCHAR2);

This procedure migrates a queue table from 8.0, 8.1, or 10.0 to 8.0, 8.1, or 10.0. Only the owner of the queue table can migrate it.

If a schema was created by an import of an export dump from a lower release or has Oracle Streams Advanced Queuing queues upgraded from a lower release, then attempts to drop it with DROP USER CASCADE will fail with ORA-24005. To drop such schemas:

1. Event 10851 should be set to level 1.

2. Drop all tables of the form AQ$_queue_table_name_NR from the schema.

3. Turn off event 10851.

4. Drop the schema.

BEGIN
   DBMS_AQADM.MIGRATE_QUEUE_TABLE (
           queue_table    => 'test.xml_qtab', 
           compatible     => '10.0');
END;
/

Creating a Queue

DBMS_AQADM.CREATE_QUEUE(
   queue_name          IN       VARCHAR2,
   queue_table         IN       VARCHAR2,
   queue_type          IN       BINARY_INTEGER DEFAULT NORMAL_QUEUE,
   max_retries         IN       NUMBER         DEFAULT NULL,
   retry_delay         IN       NUMBER         DEFAULT 0,
   retention_time      IN       NUMBER         DEFAULT 0,
   dependency_tracking IN       BOOLEAN        DEFAULT FALSE,
   comment             IN       VARCHAR2       DEFAULT NULL,

This procedure creates a queue. It has the following parameters:

All queue names must be unique within a schema. Once a queue is created with CREATE_QUEUE, it can be enabled by calling START_QUEUE. By default, the queue is created with both enqueue and dequeue disabled. To view retained messages, you can either dequeue by message ID or use SQL. If GLOBAL_TOPIC_ENABLED = TRUE when a queue is created, then a corresponding LDAP entry is also created.

The following examples (Example 8-23 through Example 8-30) use data structures created in Example 8-1 through Example 8-12.

BEGIN
   DBMS_AQADM.CREATE_QUEUE(
      queue_name         =>  'test.obj_queue',
      queue_table        =>  'test.obj_qtab');
END;
/

BEGIN
   DBMS_AQADM.CREATE_QUEUE( 
      queue_name          => 'test.raw_queue', 
      queue_table         => 'test.raw_qtab'); 
END;
/

BEGIN
   DBMS_AQADM.CREATE_QUEUE( 
      queue_name          => 'test.lob_queue', 
      queue_table         => 'test.lob_qtab'); 
END;
/

BEGIN
   DBMS_AQADM.CREATE_QUEUE( 
      queue_name          => 'test.group_queue', 
      queue_table         => 'test.group_qtab'); 
END;
/

BEGIN
   DBMS_AQADM.CREATE_QUEUE(
      queue_name         => 'test.priority_queue', 
      queue_table        => 'test.priority_qtab');
END;
/

BEGIN
   DBMS_AQADM.CREATE_QUEUE(
      queue_name         => 'test.multiconsumer_queue',
      queue_table        => 'test.multiconsumer_qtab');
END;
/

BEGIN
   DBMS_AQADM.CREATE_QUEUE(
      queue_name        => 'test.another_queue',
      queue_table       => 'test.multiconsumer_qtab');
END;
/

BEGIN
   DBMS_AQADM.CREATE_QUEUE( 
      queue_name         => 'test.multiconsumer_81_queue', 
      queue_table        => 'test.multiconsumer_81_qtab'); 
END;
/

Altering a Queue

DBMS_AQADM.ALTER_QUEUE(
   queue_name        IN    VARCHAR2,
   max_retries       IN    NUMBER   DEFAULT NULL,
   retry_delay       IN    NUMBER   DEFAULT NULL,
   retention_time    IN    NUMBER   DEFAULT NULL,
   comment           IN    VARCHAR2 DEFAULT NULL);

This procedure alters existing properties of a queue.

Only max_retries, comment, retry_delay, and retention_time can be altered. To view retained messages, you can either dequeue by message ID or use SQL. If GLOBAL_TOPIC_ENABLED = TRUE when a queue is modified, then a corresponding LDAP entry is also altered.

Example 8-31 changes retention time, saving messages for 1 day after dequeuing.

BEGIN
   DBMS_AQADM.ALTER_QUEUE( 
      queue_name        => 'test.another_queue', 
      retention_time    => 86400);
END;
/

Starting a Queue

DBMS_AQADM.START_QUEUE( 
   queue_name      IN     VARCHAR2,
   enqueue         IN     BOOLEAN DEFAULT TRUE,
   dequeue         IN     BOOLEAN DEFAULT TRUE);

This procedure enables the specified queue for enqueuing or dequeuing.

After creating a queue, the administrator must use START_QUEUE to enable the queue. The default is to enable it for both enqueue and dequeue. Only dequeue operations are allowed on an exception queue. This operation takes effect when the call completes and does not have any transactional characteristics.

BEGIN
   DBMS_AQADM.START_QUEUE (
      queue_name         => 'test.obj_queue');
END;
/

BEGIN
   DBMS_AQADM.START_QUEUE( 
      queue_name         => 'test.raw_queue', 
      dequeue            => TRUE, 
      enqueue            => FALSE); 
END;
/

Stopping a Queue

DBMS_AQADM.STOP_QUEUE(
   queue_name      IN   VARCHAR2,
   enqueue         IN   BOOLEAN DEFAULT TRUE,
   dequeue         IN   BOOLEAN DEFAULT TRUE,
   wait            IN   BOOLEAN DEFAULT TRUE);

This procedure disables enqueuing, dequeuing, or both on the specified queue.

By default, this call disables both enqueue and dequeue. A queue cannot be stopped if there are outstanding transactions against the queue. This operation takes effect when the call completes and does not have any transactional characteristics.

BEGIN
   DBMS_AQADM.STOP_QUEUE(
      queue_name        => 'test.obj_queue');
END;
/

Dropping a Queue

DBMS_AQADM.DROP_QUEUE(
   queue_name        IN    VARCHAR2,

This procedure drops an existing queue. DROP_QUEUE is not allowed unless STOP_QUEUE has been called to disable the queue for both enqueuing and dequeuing. All the queue data is deleted as part of the drop operation.

If GLOBAL_TOPIC_ENABLED = TRUE when a queue is dropped, then a corresponding LDAP entry is also dropped.

BEGIN
   DBMS_AQADM.DROP_QUEUE(
      queue_name         => 'test.obj_queue');
END;
/

Creating a Transformation

DBMS_TRANSFORM.CREATE_TRANSFORMATION(
    schema               VARCHAR2(30),
    name                 VARCHAR2(30),
    from_schema          VARCHAR2(30),
    from_type            VARCHAR2(30),
    to_schema            VARCHAR2(30),
    to_type              VARCHAR2(30),
    transformation       VARCHAR2(4000)); 

This procedure creates a message format transformation. The transformation must be a SQL function with input type from_type, returning an object of type to_type. It can also be a SQL expression of type to_type, referring to from_type. All references to from_type must be of the form source.user_data.

You must be granted EXECUTE privilege on dbms_transform to use this feature. This privilege is included in the AQ_ADMINISTRATOR_ROLE.

You must also have EXECUTE privilege on the user-defined types that are the source and destination types of the transformation, and have EXECUTE privileges on any PL/SQL function being used in the transformation function. The transformation cannot write the database state (that is, perform DML operations) or commit or rollback the current transaction.

BEGIN
   DBMS_TRANSFORM.CREATE_TRANSFORMATION(
      schema         => 'test',
      name           => 'message_order_transform', 
      from_schema    => 'test',
      from_type      => 'message_typ', 
      to_schema      => 'test',
      to_type        => 'order_typ',
      transformation => 'test.order_typ(
         source.user_data.sender_id,
         source.user_data.subject,
         source.user_data.text)');
END;
/

Modifying a Transformation

DBMS_TRANSFORM.MODIFY_TRANSFORMATION(
    schema            VARCHAR2(30),
    name              VARCHAR2(30),
    attribute_number  INTEGER,
    transformation    VARCHAR2(4000));

This procedure changes the transformation function and specifies transformations for each attribute of the target type. If the attribute number 0 is specified, then the transformation expression singularly defines the transformation from the source to target types.

All references to from_type must be of the form source.user_data. All references to the attributes of the source type must be prefixed by source.user_data.

You must be granted EXECUTE privileges on dbms_transform to use this feature. You must also have EXECUTE privileges on the user-defined types that are the source and destination types of the transformation, and have EXECUTE privileges on any PL/SQL function being used in the transformation function.

Dropping a Transformation

DBMS_TRANSFORM.DROP_TRANSFORMATION (
    schema      VARCHAR2(30),
    name        VARCHAR2(30));

This procedure drops a transformation.

You must be granted EXECUTE privileges on dbms_transform to use this feature. You must also have EXECUTE privileges on the user-defined types that are the source and destination types of the transformation, and have EXECUTE privileges on any PL/SQL function being used in the transformation function.

Granting Oracle Streams Advanced Queuing System Privileges

DBMS_AQADM.GRANT_SYSTEM_PRIVILEGE(
   privilege         IN    VARCHAR2,
   grantee           IN    VARCHAR2,
   admin_option      IN    BOOLEAN := FALSE);

This procedure grants Oracle Streams Advanced Queuing system privileges to users and roles. The privileges are ENQUEUE_ANY, DEQUEUE_ANY, MANAGE_ANY. Initially, only SYS and SYSTEM can use this procedure successfully.

Users granted the ENQUEUE_ANY privilege are allowed to enqueue messages to any queues in the database. Users granted the DEQUEUE_ANY privilege are allowed to dequeue messages from any queues in the database. Users granted the MANAGE_ANY privilege are allowed to run DBMS_AQADM calls on any schemas in the database.

BEGIN
   DBMS_AQADM.GRANT_SYSTEM_PRIVILEGE(
      privilege          =>    'ENQUEUE_ANY', 
      grantee            =>    'test', 
      admin_option       =>     FALSE);
   DBMS_AQADM.GRANT_SYSTEM_PRIVILEGE(
      privilege          =>     'DEQUEUE_ANY', 
      grantee            =>     'test', 
      admin_option       =>      FALSE);
END;
/

Revoking Oracle Streams Advanced Queuing System Privileges

DBMS_AQADM.REVOKE_SYSTEM_PRIVILEGE(
   privilege         IN   VARCHAR2,
   grantee           IN   VARCHAR2);

This procedure revokes Oracle Streams Advanced Queuing system privileges from users and roles. The privileges are ENQUEUE_ANY, DEQUEUE_ANY and MANAGE_ANY. The ADMIN option for a system privilege cannot be selectively revoked.

Users granted the ENQUEUE_ANY privilege are allowed to enqueue messages to any queues in the database. Users granted the DEQUEUE_ANY privilege are allowed to dequeue messages from any queues in the database. Users granted the MANAGE_ANY privilege are allowed to run DBMS_AQADM calls on any schemas in the database.

BEGIN
   DBMS_AQADM.REVOKE_SYSTEM_PRIVILEGE(
      privilege          =>    'DEQUEUE_ANY', 
      grantee            =>    'test');
END;
/

Granting Queue Privileges

DBMS_AQADM.GRANT_QUEUE_PRIVILEGE(
   privilege        IN    VARCHAR2,
   queue_name       IN    VARCHAR2,
   grantee          IN    VARCHAR2,
   grant_option     IN    BOOLEAN := FALSE);

This procedure grants privileges on a queue to users and roles. The privileges are ENQUEUE, DEQUEUE, or ALL. Initially, only the queue table owner can use this procedure to grant privileges on the queues.

BEGIN
   DBMS_AQADM.GRANT_QUEUE_PRIVILEGE (
      privilege     =>     'ALL', 
      queue_name    =>     'test.multiconsumer_81_queue',
      grantee       =>     'test_adm', 
      grant_option  =>      TRUE);
END;
/

Revoking Queue Privileges

DBMS_AQADM.REVOKE_QUEUE_PRIVILEGE (
   privilege         IN      VARCHAR2,
   queue_name        IN      VARCHAR2,
   grantee           IN      VARCHAR2);

This procedure revokes privileges on a queue from users and roles. The privileges are ENQUEUE or DEQUEUE.

To revoke a privilege, the revoker must be the original grantor of the privilege. The privileges propagated through the GRANT option are revoked if the grantor's privileges are revoked.

You can revoke the dequeue right of a grantee on a specific queue, leaving the grantee with only the enqueue right as in Example 8-40.

BEGIN
   DBMS_AQADM.REVOKE_QUEUE_PRIVILEGE(
      privilege     =>     'DEQUEUE', 
      queue_name    =>     'test.multiconsumer_81_queue', 
      grantee       =>     'test_adm');
END;

Adding a Subscriber

DBMS_AQADM.ADD_SUBSCRIBER (
   queue_name     IN    VARCHAR2,
   subscriber     IN    sys.aq$_agent,
   rule           IN    VARCHAR2 DEFAULT NULL,
   transformation IN    VARCHAR2 DEFAULT NULL,
   queue_to_queue IN    BOOLEAN DEFAULT FALSE,
   delivery_mode  IN    PLS_INTEGER DEFAULT PERSISTENT);

This procedure adds a default subscriber to a queue.

An application can enqueue messages to a specific list of recipients or to the default list of subscribers. This operation succeeds only on queues that allow multiple consumers, and the total number of subscribers must be 1024 or less. This operation takes effect immediately and the containing transaction is committed. Enqueue requests that are executed after the completion of this call reflect the new action. Any string within the rule must be quoted (with single quotation marks) as follows:

rule   => 'PRIORITY <= 3 AND CORRID =  ''FROM JAPAN'''

User data properties or attributes apply only to object payloads and must be prefixed with tab.userdata in all cases.

If GLOBAL_TOPIC_ENABLED is set to true when a subscriber is created, then a corresponding LDAP entry is also created.

Specify the name of the transformation to be applied during dequeue or propagation. The transformation must be created using the DBMS_TRANSFORM package.

For queues that contain payloads with XMLType attributes, you can specify rules that contain operators such as XMLType.existsNode() and XMLType.extract().

If parameter queue_to_queue is set to TRUE, then the added subscriber is a queue-to-queue subscriber. When queue-to-queue propagation is set up between a source queue and a destination queue, queue-to-queue subscribers receive messages through that propagation schedule.

If the delivery_mode parameter is the default PERSISTENT, then the subscriber receives only persistent messages. If it is set to BUFFERED, then the subscriber receives only buffered messages. If it is set to PERSISTENT_OR_BUFFERED, then the subscriber receives both types. You cannot alter this parameter with ALTER_SUBSCRIBER.

The agent name should be NULL if the destination queue is a single consumer queue.

DECLARE 
   subscriber          sys.aq$_agent; 
BEGIN 
   subscriber := sys.aq$_agent('subscriber1', 'test2.msg_queue2@london', null); 
   DBMS_AQADM.ADD_SUBSCRIBER(
      queue_name         => 'test.multiconsumer_81_queue', 
      subscriber         =>  subscriber); 
END;
/

DECLARE 
   subscriber          sys.aq$_agent; 
BEGIN 
   subscriber := sys.aq$_agent('subscriber1', 'test2.msg_queue2@london', null); 
   DBMS_AQADM.ADD_SUBSCRIBER(
      queue_name         => 'test.multiconsumer_81_queue', 
      subscriber         =>  subscriber); 
END;
/

DECLARE 
   subscriber       sys.aq$_agent; 
BEGIN 
   subscriber := sys.aq$_agent('subscriber2', 'test2.msg_queue2@london', null); 
   DBMS_AQADM.ADD_SUBSCRIBER(
      queue_name =>  'test.multiconsumer_81_queue', 
      subscriber =>   subscriber, 
      rule       =>  'priority < 2'); 
END;
/

DECLARE 
   subscriber       sys.aq$_agent; 
BEGIN 
   subscriber := sys.aq$_agent('subscriber3', 'test2.msg_queue2@london', null); 
   DBMS_AQADM.ADD_SUBSCRIBER(
      queue_name     => 'test.multiconsumer_81_queue', 
      subscriber     =>  subscriber, 
      transformation => 'test.message_order_transform');
END;
/

DECLARE 
   subscriber          SYS.AQ$_AGENT; 
BEGIN 
  subscriber := SYS.AQ$_AGENT(NULL, 'test2.single_consumer__queue@london', 
null); 
  DBMS_AQADM.ADD_SUBSCRIBER( 
      queue_name         => 'test.multiconsumer_81_queue', 
      subscriber         =>  subscriber); 
END; 

Altering a Subscriber

DBMS_AQADM.ALTER_SUBSCRIBER (
   queue_name     IN    VARCHAR2,
   subscriber     IN    sys.aq$_agent,
   rule           IN    VARCHAR2
   transformation IN    VARCHAR2);

This procedure alters existing properties of a subscriber to a specified queue.

The rule, the transformation, or both can be altered. If you alter only one of these attributes, then specify the existing value of the other attribute to the alter call. If GLOBAL_TOPIC_ENABLED = TRUE when a subscriber is modified, then a corresponding LDAP entry is created.

DECLARE 
   subscriber       sys.aq$_agent; 
BEGIN 
   subscriber := sys.aq$_agent('subscriber2', 'test2.msg_queue2@london', null); 
   DBMS_AQADM.ALTER_SUBSCRIBER(
      queue_name => 'test.multiconsumer_81_queue', 
      subscriber =>  subscriber, 
      rule       => 'priority = 1'); 
END;
/

Removing a Subscriber

DBMS_AQADM.REMOVE_SUBSCRIBER (
   queue_name         IN         VARCHAR2,
   subscriber         IN         sys.aq$_agent);

This procedure removes a default subscriber from a queue.

This operation takes effect immediately and the containing transaction is committed. All references to the subscriber in existing messages are removed as part of the operation. If GLOBAL_TOPIC_ENABLED = TRUE when a subscriber is dropped, then a corresponding LDAP entry is also dropped.

It is not an error to run the REMOVE_SUBSCRIBER procedure even when there are pending messages that are available for dequeue by the consumer. These messages are automatically made unavailable for dequeue when the REMOVE_SUBSCRIBER procedure finishes.

DECLARE
   subscriber       sys.aq$_agent;
BEGIN
   subscriber := sys.aq$_agent ('subscriber2', 'test2.msg_queue2@london', null);
   DBMS_AQADM.REMOVE_SUBSCRIBER(
      queue_name => 'test.multiconsumer_81_queue',
      subscriber => subscriber);
END;
/

Scheduling a Queue Propagation

DBMS_AQADM.SCHEDULE_PROPAGATION (
   queue_name         IN  VARCHAR2,
   destination        IN  VARCHAR2 DEFAULT NULL,
   start_time         IN  DATE     DEFAULT SYSDATE,
   duration           IN  NUMBER   DEFAULT NULL,
   next_time          IN  VARCHAR2 DEFAULT NULL,
   latency            IN  NUMBER   DEFAULT 60,
   destination_queue  IN  VARCHAR2 DEFAULT NULL);

This procedure schedules propagation of messages.

The destination can be identified by a database link in the destination parameter, a queue name in the destination_queue parameter, or both. Specifying only a database link results in queue-to-dblink propagation. If you propagate messages to several queues in another database, then all propagations have the same frequency.

If a private database link in the schema of the queue table owner has the same name as a public database link, AQ always uses the private database link.

Specifying the destination queue name results in queue-to-queue propagation was introduce in Oracle Streams Advanced Queuing 10g Release 2 (10.2). If you propagate messages to several queues in another database, queue-to-queue propagation enables you to configure each schedule independently of the others. You can enable or disable individual propagations.

Queue-to-queue propagation mode supports transparent failover when propagating to a destination Oracle Real Application Clusters (Oracle RAC) system. With queue-to-queue propagation, it is not required to repoint a database link if the owner instance of the queue fails on Oracle RAC.

Messages can also be propagated to other queues in the same database by specifying a NULL destination. If a message has multiple recipients at the same destination in either the same or different queues, then the message is propagated to all of them at the same time.

The source queue must be in a queue table meant for multiple consumers. If you specify a single-consumer queue, than error ORA-24039 results. Oracle Streams Advanced Queuing does not support the use of synonyms to refer to queues or database links.

If you specify a propagation next_time and duration, propagation will run periodically for the specified duration.If you specify a latency of zero with no next_time or duration, the resulting propagation will run forever, propagating messages as they appear in the queue, and idling otherwise. If a non-zero latency is specified, with no next_time or duration (default), the propagation schedule will be event-based. It will be scheduled to run when there are messages in the queue to be propagated. When there are no more messages for a system-defined period of time, the job will stop running until there are new messages to be propagated.The time at which the job runs depends on other factors, such as the number of ready jobs and the number of job queue processes.

Propagation uses a linear backoff scheme for retrying propagation from a schedule that encountered a failure. If a schedule continuously encounters failures, then the first retry happens after 30 seconds, the second after 60 seconds, the third after 120 seconds and so forth. If the retry time is beyond the expiration time of the current window, then the next retry is attempted at the start time of the next window. A maximum of 16 retry attempts are made after which the schedule is automatically disabled.

If you specify a value for destination that does not exist, then this procedure still runs without throwing an error. You can query runtime propagation errors in the LAST_ERROR_MSG column of the USER_QUEUE_SCHEDULES view.

BEGIN
   DBMS_AQADM.SCHEDULE_PROPAGATION(
      queue_name    =>    'test.multiconsumer_queue');
END;
/

BEGIN
   DBMS_AQADM.SCHEDULE_PROPAGATION(
      queue_name    =>    'test.multiconsumer_queue', 
      destination   =>    'another_db.world');
END;
/

BEGIN
   DBMS_AQADM.SCHEDULE_PROPAGATION(
      queue_name         =>    'test.multiconsumer_queue', 
      destination        =>    'another_db.world'
      destination_queue  =>    'target_queue');
END;
/

Verifying Propagation Queue Type

DBMS_AQADM.VERIFY_QUEUE_TYPES(
   src_queue_name    IN    VARCHAR2,
   dest_queue_name   IN    VARCHAR2,
   destination       IN    VARCHAR2 DEFAULT NULL,
   rc                OUT   BINARY_INTEGER);

This procedure verifies that the source and destination queues have identical types. The result of the verification is stored in the dictionary table SYS.AQ$_MESSAGE_TYPES, overwriting all previous output of this command.

If the source and destination queues do not have identical types and a transformation was specified, then the transformation must map the source queue type to the destination queue type.

SET SERVEROUTPUT ON
DECLARE 
rc      BINARY_INTEGER; 
BEGIN 
   DBMS_AQADM.VERIFY_QUEUE_TYPES(
      src_queue_name  => 'test.multiconsumer_queue', 
      dest_queue_name => 'test.another_queue',
      rc              =>  rc); 
   DBMS_OUTPUT.PUT_LINE('Compatible: '||rc);
END;
/

Example 8-51 involves two queues of the same type. It returns:

VQT: new style queue
Compatible: 1

If the same example is run with test.raw_queue (a queue of type RAW) in place of test.another_queue, then it returns:

VQT: new style queue
Compatible: 0

Altering a Propagation Schedule

DBMS_AQADM.ALTER_PROPAGATION_SCHEDULE( 
   queue_name         IN  VARCHAR2, 
   destination        IN  VARCHAR2 DEFAULT NULL,
   duration           IN  NUMBER   DEFAULT NULL, 
   next_time          IN  VARCHAR2 DEFAULT NULL, 
   latency            IN  NUMBER   DEFAULT 60,
   destination_queue  IN  VARCHAR2 DEFAULT NULL);

This procedure alters parameters for a propagation schedule. The destination_queue parameter for queue-to-queue propagation cannot be altered.

BEGIN
   DBMS_AQADM.ALTER_PROPAGATION_SCHEDULE(
      queue_name    =>    'test.multiconsumer_queue', 
      duration      =>    '2000', 
      next_time     =>    'SYSDATE + 3600/86400',
      latency       =>    '32'); 
END;
/

BEGIN
   DBMS_AQADM.ALTER_PROPAGATION_SCHEDULE(
      queue_name    =>    'test.multiconsumer_queue', 
      destination   =>    'another_db.world', 
      duration      =>    '2000', 
      next_time     =>    'SYSDATE + 3600/86400',
      latency       =>    '32');
END;
/

Enabling a Propagation Schedule

DBMS_AQADM.ENABLE_PROPAGATION_SCHEDULE( 
   queue_name        IN   VARCHAR2, 
   destination       IN   VARCHAR2 DEFAULT NULL,
   destination_queue IN   VARCHAR2 DEFAULT NULL);

This procedure enables a previously disabled propagation schedule.

BEGIN
   DBMS_AQADM.ENABLE_PROPAGATION_SCHEDULE(
      queue_name   =>   'test.multiconsumer_queue');
END;
/

BEGIN
   DBMS_AQADM.ENABLE_PROPAGATION_SCHEDULE(
      queue_name    =>    'test.multiconsumer_queue', 
      destination   =>    'another_db.world');
END;
/

Disabling a Propagation Schedule

DBMS_AQADM.DISABLE_PROPAGATION_SCHEDULE( 
   queue_name        IN   VARCHAR2, 
   destination       IN   VARCHAR2 DEFAULT NULL,
   destination_queue IN   VARCHAR2 DEFAULT NULL);

This procedure disables a previously enabled propagation schedule.

BEGIN
   DBMS_AQADM.DISABLE_PROPAGATION_SCHEDULE(
      queue_name   =>   'test.multiconsumer_queue');
END;
/

BEGIN
   DBMS_AQADM.DISABLE_PROPAGATION_SCHEDULE(
      queue_name    =>    'test.multiconsumer_queue', 
      destination   =>    'another_db.world');
END;
/

Unscheduling a Queue Propagation

DBMS_AQADM.UNSCHEDULE_PROPAGATION (
   queue_name        IN  VARCHAR2,
   destination       IN  VARCHAR2 DEFAULT NULL,
   destination_queue IN  VARCHAR2 DEFAULT NULL);

This procedure unschedules a previously scheduled propagation of messages from a queue to a destination. The destination is identified by a specific database link in the destination parameter or by name in the destination_queue parameter.

BEGIN
   DBMS_AQADM.UNSCHEDULE_PROPAGATION(
      queue_name => 'test.multiconsumer_queue'); 
END;
/

BEGIN
   DBMS_AQADM.UNSCHEDULE_PROPAGATION(
      queue_name    =>   'test.multiconsumer_queue', 
      destination   =>   'another_db.world');
END;
/

Creating an Oracle Streams Advanced Queuing Agent

DBMS_AQADM.CREATE_AQ_AGENT (
  agent_name                IN VARCHAR2,
  certificate_location      IN VARCHAR2 DEFAULT NULL,
  enable_http               IN BOOLEAN DEFAULT FALSE,
  enable_anyp               IN BOOLEAN DEFAULT FALSE);

This procedure registers an agent for Oracle Streams Advanced Queuing Internet access using HTTP protocols.

The SYS.AQ$INTERNET_USERS view has a list of all Oracle Streams Advanced Queuing Internet agents. When an agent is created, altered, or dropped, an LDAP entry is created for the agent if the following are true:

• GLOBAL_TOPIC_ENABLED = TRUE

• certificate_location is specified

Altering an Oracle Streams Advanced Queuing Agent

DBMS_AQADM.ALTER_AQ_AGENT (
  agent_name                IN VARCHAR2,
  certificate_location      IN VARCHAR2 DEFAULT NULL,
  enable_http               IN BOOLEAN DEFAULT FALSE,
  enable_anyp               IN BOOLEAN DEFAULT FALSE);

This procedure alters an agent registered for Oracle Streams Advanced Queuing Internet access.

When an Oracle Streams Advanced Queuing agent is created, altered, or dropped, an LDAP entry is created for the agent if the following are true:

• GLOBAL_TOPIC_ENABLED = TRUE

• certificate_location is specified

Dropping an Oracle Streams Advanced Queuing Agent

DBMS_AQADM.DROP_AQ_AGENT (
  agent_name     IN VARCHAR2);

This procedure drops an agent that was previously registered for Oracle Streams Advanced Queuing Internet access.

When an Oracle Streams Advanced Queuing agent is created, altered, or dropped, an LDAP entry is created for the agent if the following are true:

• GLOBAL_TOPIC_ENABLED = TRUE

• certificate_location is specified

Enabling Database Access

DBMS_AQADM.ENABLE_DB_ACCESS (
  agent_name                IN VARCHAR2,
  db_username               IN VARCHAR2)

This procedure grants an Oracle Streams Advanced Queuing Internet agent the privileges of a specific database user. The agent should have been previously created using the CREATE_AQ_AGENT procedure.

The SYS.AQ$INTERNET_USERS view has a list of all Oracle Streams Advanced Queuing Internet agents and the names of the database users whose privileges are granted to them.

Disabling Database Access

DBMS_AQADM.DISABLE_DB_ACCESS (
  agent_name                IN VARCHAR2,
  db_username               IN VARCHAR2)

This procedure revokes the privileges of a specific database user from an Oracle Streams Advanced Queuing Internet agent. The agent should have been previously granted those privileges using the ENABLE_DB_ACCESS procedure.