==============================================================
Using the payload of an AQ as sys.AQ$_JMS_TEXT_MESSAGE type is more
effective instead of creating a custom object type. specially when the source
system is enqueueing data using java.
*Create
roles and grant privileges:
CREATE ROLE my_aq_adm_role;
GRANT CONNECT, RESOURCE,
aq_administrator_role TO my_aq_adm_role;
--
CREATE ROLE my_aq_user_role;
GRANT CREATE SESSION,
aq_user_role TO
my_aq_user_role;
BEGIN
DBMS_AQADM.GRANT_SYSTEM_PRIVILEGE('ENQUEUE_ANY','my_aq_user_role',FALSE);
DBMS_AQADM.GRANT_SYSTEM_PRIVILEGE('DEQUEUE_ANY','my_aq_user_role',FALSE);
END;
*Create
users, assign roles and grant privileges:
CREATE USER aqadm IDENTIFIED BY aqadm DEFAULT TABLESPACE <tablespace
name> TEMPORARY TABLESPACE temp;
GRANT my_aq_adm_role TO aqadm;
CREATE USER aquser IDENTIFIED BY aquser DEFAULT TABLESPACE <tablespace
name> TEMPORARY TABLESPACE temp;
GRANT my_aq_user_role TO aquser;
*All
the required grants:
GRANT EXECUTE ON
dbms_aq TO aquser;
GRANT RESOURCE TO aquser;
GRANT CONNECT TO aquser;
GRANT EXECUTE ANY PROCEDURE TO aquser;
GRANT EXECUTE ON dbms_aqadm TO aquser;
GRANT EXECUTE ON dbms_aq TO aquser;
GRANT EXECUTE ON dbms_aqin TO aquser;
*Login
as ‘aqadm’ and create Queue table and Queue:
DECLARE
v_q_table VARCHAR2(25) := 'xx_test_ib_aq_table';
v_q_name VARCHAR2(20) := 'xx_test_inbound_aq';
v_pl_type VARCHAR2(25) := 'sys.AQ$_JMS_TEXT_MESSAGE';
BEGIN
DBMS_AQADM.CREATE_QUEUE_TABLE (queue_table => v_q_table,
queue_payload_type => v_pl_type,
multiple_consumers => FALSE
);
DBMS_AQADM.CREATE_QUEUE (queue_name => v_q_name,
queue_table =>
v_q_table
);
DBMS_AQADM.START_QUEUE (
queue_name =>
v_q_name
);
END;
/
*Connect
to ‘aquser’ and Create a temporary table for testing:
DROP TABLE xx_test_dq_data_tbl
/
CREATE TABLE xx_test_dq_data_tbl
(message_text_id VARCHAR2(50)
,message_data
VARCHAR2(4000)
)
/
*Create
CALL BACK procedure:
CREATE OR REPLACE PACKAGE xx_temp_callback_dq_pkg
AS
PROCEDURE dequeue_p (context IN RAW
,reginfo IN SYS.aq$_reg_info
,descr IN SYS.aq$_descriptor
,payload IN RAW
,payloadl IN NUMBER
);
END xx_temp_callback_dq_pkg;
/
CREATE OR REPLACE PACKAGE BODY xx_temp_callback_dq_pkg
AS
PROCEDURE dequeue_p (context IN RAW
,reginfo IN SYS.aq$_reg_info
,descr IN SYS.aq$_descriptor
,payload IN RAW
,payloadl IN NUMBER
)
AS
v_dequeue_options DBMS_AQ.DEQUEUE_OPTIONS_T;
v_message_properties DBMS_AQ.MESSAGE_PROPERTIES_T;
v_msgid RAW(100);
my_message sys.AQ$_JMS_TEXT_MESSAGE;
v_msg_id_text VARCHAR2(15) := 'MESSAGE_TYPE_ID';
v_message_text_id VARCHAR2(50);
v_message_data VARCHAR2(4000);
BEGIN
v_dequeue_options.msgid := descr.msg_id;
v_dequeue_options.wait := DBMS_AQ.no_wait;
DBMS_AQ.DEQUEUE(queue_name
=> 'aqadm.xx_test_ib_aq_table'
,dequeue_options =>
v_dequeue_options
,message_properties
=>
v_message_properties
,payload => my_message
,msgid => v_msgid
);
v_message_text_id := my_message.get_string_property('MESSAGE_TYPE_ID');
my_message.get_text(v_message_data);
INSERT INTO xx_test_dq_data_tbl
(message_text_id
,message_data
)
VALUES (v_message_text_id
,v_message_data
);
COMMIT;
END dequeue_p;
END xx_temp_callback_dq_pkg;
/
*Register
the Call Back procedure:
BEGIN
DBMS_AQ.REGISTER (SYS.AQ$_REG_INFO_LIST
(SYS.AQ$_REG_INFO
('xx_test_ib_aq_table
,DBMS_AQ.NAMESPACE_AQ
,'plsql://xx_temp_callback_dq_pkg.dequeue_p'
,HEXTORAW('FF')
)
)
,1
);
END;
/
*Now
to test the call back procedure, Enqueue a message in to the AQ:
DECLARE
enq_msgid
raw(16);
eopt dbms_aq.enqueue_options_t;
mprop
dbms_aq.message_properties_t;
l_msg_type
sys.AQ$_JMS_TEXT_MESSAGE
;
BEGIN
l_msg_type := sys.aq$_jms_text_message.construct;
l_msg_type.set_string_property('MESSAGE_TYPE_ID', 'TEST_MESSAGE_FROM_JOSEPH');
l_msg_type.set_text('01000111 01001111 01000100
00100000 01010011 01001111 00100000 01001100 01001111 01010110 01000101
01000100 00100000 01010100 01001000 01000101 00100000 01010111 01001111
01010010 01001100 01000100') ;
DBMS_AQ.ENQUEUE(queue_name
=> 'aqadm.xx_test_ib_aq_table',
enqueue_options => eopt,
message_properties => mprop,
payload => l_msg_type,
msgid => enq_msgid );
END;
/
* Query the
temporary table to see that the enqueued message is automatically dequeued and
inserted into xx_test_dq_data_tbl.
==============================================================
