Example for using PLSQL Advanced Queue through CALL BACK procedure - Payload : “sys.AQ$_JMS_TEXT_MESSAGE” type

==============================================================

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.

==============================================================