GRANT AQ_ADMINISTRATOR_ROLE TO Cust
/
grant EXECUTE on DBMS_AQ to Cust
/
GRANT EXECUTE ON DBMS_AQADM to Cust
/
GRANT EXECUTE ON dbms_aq TO PUBLIC
/
EXECUTE dbms_aqadm.grant_system_privilege('ENQUEUE_ANY','Cust',TRUE);
EXECUTE dbms_aqadm.grant_system_privilege('DEQUEUE_ANY','Cust',TRUE);

EXECUTE dbms_aqadm.grant_system_privilege('ENQUEUE_ANY','cust_upd',FALSE);
EXECUTE dbms_aqadm.grant_system_privilege('ENQUEUE_ANY','PUBLIC',FALSE);


REM =====================================================
[b]REM Create type CREATE_queue_table, CREATE_queue, start_queue[/b]
REM =====================================================

CREATE type msg_q_type
AS Object
  (
STORE_KEY                       NUMBER(5) ,                                                                                                                                                                                    
CUST_NO                         NUMBER(12)    ,                                                                                                                                                                               
TABLE_NM                     VARCHAR2(10 ))
;
/


[b]--Create Q table[/b]
BEGIN
  dbms_aqadm.CREATE_queue_table
          ( queue_table => 'msg_q_tbl', 
            --sort_list => 'priority',
            queue_payload_type => 'msg_q_type',
            multiple_consumers => TRUE,              
            COMMENT => 'Creating input queue table');
END;
/
[b]--My Queue[/b]
BEGIN
  dbms_aqadm.CREATE_queue( queue_name => 'msg_queue', queue_table => 'msg_q_tbl', COMMENT => 'Test Queue');
END;
/


BEGIN
  [b]dbms_aqadm.start_queue( queue_name => 'msg_queue');[/b]
END;
/


REM ==============================================
REM SETUP complete 
REM ==============================================

[b]PROMPT Create procedure ENQUEUE_MSG[/b]
CREATE OR REPLACE
PROCEDURE ENQUEUE_MSG(
    pi_payload_info msg_q_type,
    pi_Owner Varchar2 Default 'CUST',
    pi_Q_Name Varchar2 Default 'msg_queue')
AS
PRAGMA AUTONOMOUS_TRANSACTION;
  eopt dbms_aq.enqueue_options_t;
  mprop dbms_aq.message_properties_t;
  enq_msgid RAW(16); 
  
  --priority NUMBER;
BEGIN
  --SELECT aq_sequence.nextval INTO priority FROM dual;
  --mprop.priority := priority;
  [i]dbms_aq.enqueue( queue_name         => pi_Owner||'.'||pi_Q_Name
                  ,enqueue_options    => eopt 
                  ,message_properties => mprop
                  ,payload            => pi_payload_info
                  ,msgid => enq_msgid);[/i]  
--Debug_MSG ('enqueue enq_msgid='||enq_msgid );     
  
  INSERT INTO CUST_MOVER_LOG ( LOG_ID , LOG_DATE, LOG_TYPE, LOG_LABEL, LOG_COMMENT )
   VALUES ( lex_seq_1.NEXTVAL
           , Systimestamp
           , 'E'
           , enq_msgid
           ,'StoreKey:' || pi_payload_info.STORE_KEY ||' CustNO:'
                           || pi_payload_info.CUST_NO||' CH:'
                           || ' TABLE_NM='||pi_payload_info.TABLE_NM
                           
            );
                        
  COMMIT;
  Exception
  	 When Others Then
  	 	 Rollback;
  	 	 Raise;
END ENQUEUE_MSG;
/



 
[b]--Call back procedure[/b]
CREATE Or Replace PROCEDURE msg_q_callback_proc(
                 context  RAW,
                 reginfo  SYS.AQ$_REG_INFO,
                 descr    SYS.AQ$_DESCRIPTOR,
                 payload  RAW,
                 payloadl NUMBER
                 ) AS

   r_dequeue_options    DBMS_AQ.DEQUEUE_OPTIONS_T;
   r_message_properties DBMS_AQ.MESSAGE_PROPERTIES_T;
   v_message_handle     RAW(16);
   o_payload            msg_q_type;

BEGIN
  --Debug_MSG ('msg_q_callback_proc running');
  
   r_dequeue_options.msgid := descr.msg_id;
   r_dequeue_options.consumer_name := descr.consumer_name;
  

   [i]DBMS_AQ.DEQUEUE(
      queue_name         => descr.queue_name,
      dequeue_options    => r_dequeue_options,
      message_properties => r_message_properties,
      payload            => o_payload,
      msgid              => v_message_handle
      );[/i] 
 Update CUST_MOVER_LOG
  Set LOG_TYPE = 'D'
  Where LOG_LABEL = v_message_handle;
 
 If sql%rowcount = 0 Then 
  INSERT INTO CUST_MOVER_LOG ( LOG_ID , LOG_DATE, LOG_TYPE, LOG_LABEL, LOG_COMMENT )
   VALUES ( lex_seq_1.NEXTVAL
           , Systimestamp
           , 'B'
           , v_message_handle
           ,'StoreKey:' || o_payload.STORE_KEY ||' CustNO:'
                           || o_payload.CUST_NO||' CH:'                          
                           || ' for table='||o_payload.TABLE_NM
            );
  End if;             
     
    /*  Debug_MSG ('msg_q_callback_proc=dequeued at [' || TO_CHAR( SYSTIMESTAMP,
                                        'DD-MON-YYYY HH24:MI:SS.FF3' ) || '] STORE_KEY-'||o_payload.STORE_KEY 
                                        ||' ORIG_SYSTEM_REFERENCE='||o_payload.ORIG_SYSTEM_REFERENCE);*/
                                        
   COMMIT;
   --Debug_MSG ('Call to check start');
 
Exception
	 When Others Then
	
	 	 Raise;	
END msg_q_callback_proc;
/

[b]--Add an agent to process requests[/b]
BEGIN

   DBMS_AQADM.ADD_SUBSCRIBER (
      queue_name => 'msg_queue',
      subscriber => SYS.AQ$_AGENT(
                       'msg_queue_subscriber',
                       NULL,
                       NULL )
      );

   [b] DBMS_AQ.REGISTER (
       SYS.AQ$_REG_INFO_LIST(
          SYS.AQ$_REG_INFO(
             'msg_queue:msg_queue_subscriber',
             DBMS_AQ.NAMESPACE_AQ,
             'plsql://msg_q_callback_proc',
             HEXTORAW('FF')
             )
          ),
       1
       );[/b]
END;
/


Create Public synonym msg_q_type for msg_q_type;
Grant all on msg_q_type to Public;
Create Public synonym ENQUEUE_MSG for ENQUEUE_MSG;
Grant execute on ENQUEUE_MSG to Public;

Prompt Altering table structure of CUST_MOVER_LOG 
Alter table CUST_MOVER_LOG Modify LOG_DATE Timestamp;
Alter table CUST_MOVER_LOG Modify LOG_ID NUMBER;

create or replace
procedure mass_update (prownum number) is
Begin

 Update customer
  set last_name = last_name||' LC' 
  where rownum < prownum;
  Commit;
End mass_update;