DECLARE
enq_cnt PLS_INTEGER;
enq_opt DBMS_AQ.enqueue_options_t;
mesg_props DBMS_AQ.message_properties_array_t;
parray OE.addr_nest_tab_typ;
msgidarr DBMS_AQ.msgid_array_t;

BEGIN

-- Add code here to populate the array --
--
enq_cnt := DBMS_AQ.ENQUEUE_ARRAY
(queue_name      => 'IX.ORDERS_QUEUE',
enqueue_options => enq_opt,
array_size      => 20,
message_properties_array => mesg_props,
payload_array   => parray,
msgid_array     => msgidarr);
END;
/

The dbms_aq  package supports programmatically interfacing with Oracle Stream Advanced Queuing, such as Queue Tables , commonly referred to as AQ. Once the AQ environment has been properly configured via dbms_sqadm, use this package to work with the actual messages held in those queues and queue tables. So the actual work is performed via this package.

There are a few enumerated constants that one must know to use this package. Also, since this package is wrapped, one cannot see the actual constant values:

VISIBILITY:                     IMMEDIATE | ON_COMMIT

DEQUEUE_MODE             BROWSE | LOCKED | REMOVE | REMOVE_NODATA

NAVIGATION:                     FIRST_MESSAGE | NEXT_MESSAGE

STATE:                   WAITING | READY | PROCESSED | EXPIRED

SEQUENCE_DEVIATION:             BEFORE | TOP

WAIT:                    FOREVER | NO_WAIT

DELAY:                   NO_DELAY

EXPIRATION:                     NONE

NAMESPACE:               NAMESPACE_AQ | NAMESPACE_ANONYMOUS

NTFN_GROUPING_CLASS:     NTFN_GROUP_TYPE_SUMMARY | NTFN_GROUPING_TYPE_LAST

NTFN_GROUPING_REPEAT_COUNT:     NTFN_GROUPING_FOREVER

Bind_agent is a procedure that creates an LDAP server entry for the Oracle AQ agent. The certificate parameter is stored on the LDAP server as an attribute (usercertificate) of the Organizational Person entity. The aq$_agent type is shown just below.

TYPE SYS.AQ$_AGENT IS OBJECT (

  name       VARCHAR2(30),

  address    VARCHAR2(1024),

  protocol   NUMBER  DEFAULT 0

);

dequeueis a procedure that simply dequeues, i.e subtracts, a message from the specified queue. The dequeue option and message property types are shown just below.

TYPE DEQUEUE_OPTIONS_T IS RECORD (

  consumer_name     VARCHAR2(30)    DEFAULT NULL,

  dequeue_mode      BINARY_INTEGER  DEFAULT REMOVE,

  navigation        BINARY_INTEGER  DEFAULT NEXT_MESSAGE,

  visibility        BINARY_INTEGER  DEFAULT ON_COMMIT,

  wait              BINARY_INTEGER  DEFAULT FOREVER,

  msgid             RAW(16)         DEFAULT NULL,

  correlation       VARCHAR2(128)   DEFAULT NULL,

  deq_condition     VARCHAR2(4000)  DEFAULT NULL,

  signature         aq$_sig_prop    DEFAULT NULL,

  transformation    VARCHAR2(61)    DEFAULT NULL,

  delivery_mode     PLS_INTEGER     DEFAULT PERSISTENT

);

TYPE message_properties_t IS RECORD (

  priority               BINARY_INTEGER  NOT NULL DEFAULT 1,

  delay                  BINARY_INTEGER  NOT NULL DEFAULT NO_DELAY,

  expiration             BINARY_INTEGER  NOT NULL DEFAULT NEVER,

  correlation            VARCHAR2(128)   DEFAULT NULL,

  attempts               BINARY_INTEGER,

  recipient_list         AQ$_RECIPIENT_LIST_T,

  exception_queue        VARCHAR2(61)    DEFAULT NULL,

  enqueue_time           DATE,

  state                  BINARY_INTEGER,

  sender_id              SYS.AQ$_AGENT   DEFAULT NULL,

  original_msgid         RAW(16)         DEFAULT NULL,

  signature              aq$_sig_prop    DEFAULT NULL,

  transaction_group      VARCHAR2(30)    DEFAULT NULL,

  user_property          SYS.ANYDATA     DEFAULT NULL

  delivery_mode          PLS_INTEGER     NOT NULL DEFAULT DBMS_AQ.PERSISTENT

);

dequeue_array  is a function that performs a bulk dequeue operation, fetching arrays of payloads, message properties and message IDs. It returns a pls_integer with the number of messages successfully dequeued. The message_properties_array_t and msgid_array_t types are shown below.

TYPE MESSAGE_PROPERTIES_ARRAY_T IS VARRAY (2147483647)

  OF MESSAGE_PROPERTIES_T;

TYPE MSGID_ARRAY_T IS TABLE OF RAW(16) INDEX BY BINARY_INTEGER

enqueue is a procedure that simply enqueues, i.e adds, a message to the specified queue. The enqueue option and message property types are shown just below.

TYPE SYS.ENQUEUE_OPTIONS_T IS RECORD (

  visibility            BINARY_INTEGER  DEFAULT ON_COMMIT,

  relative_msgid        RAW(16)         DEFAULT NULL,

  sequence_deviation    BINARY_INTEGER  DEFAULT NULL,

  transformation        VARCHAR2(61)    DEFAULT NULL,

  delivery_mode         PLS_INTEGER     NOT NULL DEFAULT PERSISTENT

);

TYPE message_properties_t IS RECORD (

  priority               BINARY_INTEGER  NOT NULL DEFAULT 1,

  delay                  BINARY_INTEGER  NOT NULL DEFAULT NO_DELAY,

  expiration             BINARY_INTEGER  NOT NULL DEFAULT NEVER,

  correlation            VARCHAR2(128)   DEFAULT NULL,

  attempts               BINARY_INTEGER,

  recipient_list         AQ$_RECIPIENT_LIST_T,

  exception_queue        VARCHAR2(61)    DEFAULT NULL,

  enqueue_time           DATE,

  state                  BINARY_INTEGER,

  sender_id              SYS.AQ$_AGENT   DEFAULT NULL,

  original_msgid         RAW(16)         DEFAULT NULL,

  signature              aq$_sig_prop    DEFAULT NULL,

  transaction_group      VARCHAR2(30)    DEFAULT NULL,

  user_property          SYS.ANYDATA     DEFAULT NULL

  delivery_mode          PLS_INTEGER     NOT NULL DEFAULT DBMS_AQ.PERSISTENT

);

enqueue_array  is a function that performs a bulk enqueue operation, posting arrays of payloads, message properties and message IDs. It returns a pls_integer with the number of messages successfully dequeued. The message_properties_array_t and msgid_array_t types are shown below.

TYPE MESSAGE_PROPERTIES_ARRAY_T IS VARRAY (2147483647)

  OF MESSAGE_PROPERTIES_T;

TYPE MSGID_ARRAY_T IS TABLE OF RAW(16) INDEX BY BINARY_INTEGER

listen is an overloaded procedure that listens for agents on one or more queues where the address field of the agent indicates the queue to monitor. It has the two forms and uses the data types shown below. The listen delivery mode can be either persistent, buffered, or resistent_buffered.

TYPE SYS.AQ$_AGENT IS OBJECT (

  name       VARCHAR2(30),

  address    VARCHAR2(1024),

  protocol   NUMBER  DEFAULT 0

);

TYPE aq$_agent_list_t IS TABLE of aq$_agent INDEXED BY BINARY_INTEGER;

TYPE aq$_agent_list_t IS TABLE of aq$_agent INDEXED BY BINARY_INTEGER;

post  is a procedure that posts anonymous subscriptions to notify any and all clients registered for notification. The aq$_post_info and aq_post_infolist types are shown next.

TYPE SYS.AQ$_POST_INFO IS OBJECT (

  name        VARCHAR2(128),

  namespace   NUMBER,

  payload     RAW(2000)  DEFAULT NULL

);

TYPE SYS.AQ$_POST_INFO_LIST AS VARRAY(1024) OF SYS.AQ$_POST_INFO;

register  is a procedure that registers for notification an email address, PL/SQL procedure or HTTP UTL. The aq$_reg_info and aq$_reg_infolist types are shown next.

TYPE SYS.AQ$_REG_INFO IS OBJECT (

  name                              VARCHAR2(128),

  namespace                         NUMBER,

  callback                          VARCHAR2(4000),

  context                           RAW(2000)  DEFAULT NULL,

  qosflags                          NUMBER,

  timeout                           NUMBER

  ntfn_grouping_class               NUMBER,

  ntfn_grouping_value               NUMBER    DEFAULT 600,

  ntfn_grouping_type                NUMBER,

  ntfn_grouping_start_time          TIMESTAMP WITH TIME ZONE,

  ntfn_grouping_repeat_count   NUMBER

);

TYPE SYS.AQ$_REG_INFO_LIST AS VARRAY(1024) OF SYS.AQ$_REG_INFO;

unbindis a procedure that simply removes form the LDAP server an AQ agent. The aq$_agent type is shown just below.

TYPE SYS.AQ$_AGENT IS OBJECT (

  name       VARCHAR2(30),

  address    VARCHAR2(1024),

  protocol   NUMBER  DEFAULT 0

);

unregister is a procedure that unregisters for notification an email address, PL/SQL procedure or HTTP UTL.

TYPE SYS.AQ$_REG_INFO IS OBJECT (

  name                              VARCHAR2(128),

  namespace                         NUMBER,

  callback                          VARCHAR2(4000),

  context                           RAW(2000)  DEFAULT NULL,

  qosflags                          NUMBER,

  timeout                           NUMBER

  ntfn_grouping_class               NUMBER,

  ntfn_grouping_value               NUMBER    DEFAULT 600,

  ntfn_grouping_type                NUMBER,

  ntfn_grouping_start_time          TIMESTAMP WITH TIME ZONE,

  ntfn_grouping_repeat_count   NUMBER

);

TYPE SYS.AQ$_REG_INFO_LIST AS VARRAY(1024) OF SYS.AQ$_REG_INFO;

If you want to understand the internals of Oracle replications, I recommend these tight-focus reference books:

• Oracle Streams Book
• Oracle networking & multi-master replication book
• Oracle PL/SQL Programming book

Also see:

• Oracle dbms_aqadm example

Advanced Oracle Utilities: The Definitive Reference by Rampant TechPress is written by top Oracle database experts (Bert Scalzo,  Donald Burleson, and Steve Callan).  Buy direct from the publisher and save 30%!