CREATE OR REPLACE
type logmessage as object (
wpTRID number(9),
wpRuleID number(9),
wpCompType Number(4),
wpComponent varchar2(40),
wpEvent varchar2(80),
wpStartTimestamp timestamp,
wpEndTimestamp timestamp,
wpSuccess varchar2(20)
)
/

DBMS_AQADM.CREATE_QUEUE (
queue_name => 'Log_Queue',
queue_table => 'Log_Table',
queue_type => DBMS_AQADM.NORMAL_QUEUE,
max_retries => 3,
comment => 'Log Queue',
auto_commit => FALSE);

PROCEDURE LOGENQUEUE(wpTRID in number, wpRuleID in number, wpCompType in number, wpComponent in varchar2, wpEvent in varchar2, wpSuccess in varchar2, wpStartTimestamp in timestamp, wpEndTimestamp in timestamp)
IS

queue_options DBMS_AQ.ENQUEUE_OPTIONS_T;
message_prop DBMS_AQ.MESSAGE_PROPERTIES_T;
message_id RAW(16);
my_msg LogMessage;
seq Number(9);

BEGIN
IF(wpTRID=-1)
THEN
select wp_queue_seq.nextval into seq from dual;
my_msg:=LogMessage(seq,wpRuleID,wpCompType,wpComponent,wpEvent, wpStartTimestamp, wpEndTimestamp, wpSuccess);
message_prop.correlation:=seq;
ELSE
my_msg:=LogMessage(wpTRID,wpRuleID,wpCompType,wpComponent,wpEvent, wpStartTimestamp, wpEndTimestamp, wpSuccess);
message_prop.correlation:=wpTRID;
END IF;

message_prop.exception_queue:='Log_ExcepQueue';

DBMS_AQ.ENQUEUE( queue_name => 'Log_Queue', enqueue_options => queue_options, message_properties => message_prop, payload => my_msg, msgid => message_id);

END LOGENQUEUE;

PROCEDURE DEQUEUE_LOG_TABLE_2
IS
options DBMS_AQ.dequeue_options_t;
properties DBMS_AQ.message_properties_t;
v_message_handle RAW(16);
log_message_a logmessage;

TYPE log_dump_a_t IS TABLE OF log_dump%ROWTYPE
INDEX BY BINARY_INTEGER;

log_dump_a log_dump_a_t;
total NUMBER := 0;
n NUMBER;
BEGIN
options.navigation := DBMS_AQ.first_message;
options.WAIT := DBMS_AQ.no_wait;

FOR rec IN (SELECT wpi_id
FROM jm_wp_info_history jm, log_table l
WHERE jm.wpi_state = 'DONE'
AND jm.wpi_id = l.user_data.wptrid
AND ROWNUM < 2
GROUP BY jm.wpi_id) LOOP
total := total + 1;
options.correlation := rec.wpi_id;

SELECT COUNT(1)
INTO n
FROM log_table
WHERE corrid = rec.wpi_id;

FOR i IN 1 .. n LOOP
DBMS_AQ.dequeue(queue_name => 'Log_Queue',
dequeue_options => options,
message_properties => properties,
payload => log_message_a,
msgid => v_message_handle
);
log_dump_a(NVL(log_dump_a.LAST + 1, 1)).wptrid := log_message_a.wptrid;
log_dump_a(log_dump_a.LAST).wpruleid := log_message_a.wpruleid;
log_dump_a(log_dump_a.LAST).wpcomptype := log_message_a.wpcomptype;
log_dump_a(log_dump_a.LAST).wpcomponent := log_message_a.wpcomponent;
log_dump_a(log_dump_a.LAST).wpevent := log_message_a.wpevent;
log_dump_a(log_dump_a.LAST).wpstarttimestamp := log_message_a.wpstarttimestamp;
log_dump_a(log_dump_a.LAST).wpendtimestamp := log_message_a.wpendtimestamp;
log_dump_a(log_dump_a.LAST).wpsuccess := log_message_a.wpsuccess;
END LOOP;


IF ( log_dump_a IS NOT NULL
AND log_dump_a.COUNT > 0) THEN
FORALL i IN log_dump_a.FIRST .. log_dump_a.LAST
INSERT INTO log_dump
VALUES log_dump_a(i);

log_dump_a.DELETE;
END IF;

COMMIT;
END LOOP;
END dequeue_log_table_2;

PROCEDURE DEQUEUE_LOG_TABLE
IS
TYPE log_message_a_t IS TABLE OF logmessage;

payloadarr log_message_a_t := log_message_a_t();
deqopt DBMS_AQ.dequeue_options_t;
msgproparr DBMS_AQ.message_properties_array_t := DBMS_AQ.message_properties_array_t();
msgidarr DBMS_AQ.msgid_array_t := DBMS_AQ.msgid_array_t();
retval PLS_INTEGER;

TYPE log_dump_a_t IS TABLE OF log_dump%ROWTYPE
INDEX BY BINARY_INTEGER;

log_dump_a log_dump_a_t;
total NUMBER := 0;
BEGIN
payloadarr.EXTEND(100);
msgproparr.EXTEND(100);


deqopt.WAIT := DBMS_AQ.no_wait;
deqopt.navigation := DBMS_AQ.first_message_multi_group;

FOR rec IN (SELECT wpi_id
FROM jm_wp_info_history jm, log_table l
WHERE jm.wpi_state = 'DONE'
AND jm.wpi_id = l.user_data.wptrid
AND ROWNUM < 2
GROUP BY jm.wpi_id) LOOP
total := total + 1;
deqopt.correlation := TO_CHAR(rec.wpi_id);

payloadarr.DELETE;
msgidarr.DELETE;
msgproparr.DELETE;
retval :=
DBMS_AQ.dequeue_array(queue_name => 'Log_Queue',
dequeue_options => deqopt,
array_size => payloadarr.COUNT,
message_properties_array => msgproparr,
payload_array => payloadarr,
msgid_array => msgidarr
);

FOR i IN 1 .. retval LOOP
log_dump_a(NVL(log_dump_a.LAST + 1, 1)).wptrid := payloadarr(i).wptrid;
log_dump_a(log_dump_a.LAST).wpruleid := payloadarr(i).wpruleid;
log_dump_a(log_dump_a.LAST).wpcomptype := payloadarr(i).wpcomptype;
log_dump_a(log_dump_a.LAST).wpcomponent := payloadarr(i).wpcomponent;
log_dump_a(log_dump_a.LAST).wpevent := payloadarr(i).wpevent;
log_dump_a(log_dump_a.LAST).wpstarttimestamp := payloadarr(i).wpstarttimestamp;
log_dump_a(log_dump_a.LAST).wpendtimestamp := payloadarr(i).wpendtimestamp;
log_dump_a(log_dump_a.LAST).wpsuccess := payloadarr(i).wpsuccess;
END LOOP;

IF ( log_dump_a IS NOT NULL
AND log_dump_a.COUNT > 0) THEN
FORALL i IN log_dump_a.FIRST .. log_dump_a.LAST
INSERT INTO log_dump
VALUES log_dump_a(i);

log_dump_a.DELETE;
END IF;

COMMIT;
END LOOP;

END dequeue_log_table;

SELECT state, COUNT(1)
FROM log_table
GROUP BY state;

State count(1)
0 1218578