Quelle est la syntaxe pour créer un abonné de file d'attente en PL/SQL?

Question

J'essaie de créer une file d'attente et un rappel qui se déclenche lorsqu'un message est mis en file d'attente, mais je n'arrive pas à déclencher le rappel. Qu'est-ce que je fais mal? J'ai un déclencheur qui met en file d'attente un message, et je peux le voir sur la table des messages de file d'attente, et je peux le déquiler à la main et le traiter, je n'arrive pas à déclencher le callback.

BEGIN  
DBMS_AQADM.CREATE_QUEUE_TABLE (
    queue_table  => 'queue_message_table', 
    queue_payload_type => 'queue_message_type', 
    multiple_consumers => TRUE); 

DBMS_AQADM.CREATE_QUEUE (
    queue_name => 'message_queue', 
    queue_table => 'queue_message_table'); 
DBMS_AQADM.START_QUEUE (queue_name => 'message_queue'); 
END; 

CREATE OR REPLACE PROCEDURE queue_callback(
    context RAW, reginfo SYS.AQ$_REG_INFO, descr SYS.AQ$_DESCRIPTOR, payload RAW, payloadl NUMBER) AS 

    queue_options  DBMS_AQ.DEQUEUE_OPTIONS_T; 
    message_properties DBMS_AQ.MESSAGE_PROPERTIES_T; 
    my_message   queue_message_type; 
    ret     varchar2(200); 
    message_id   RAW(16); 
BEGIN 
    DBMS_OUTPUT.PUT_LINE('Callback'); 
    queue_options.msgid := descr.msg_id; 
    queue_options.consumer_name := descr.consumer_name; 

    DBMS_AQ.DEQUEUE(
     queue_name => descr.queue_name, 
     dequeue_options => queue_options, 
     message_properties => message_properties, 
     payload => my_message, 
     msgid => message_id); 
    ret := handle_message(my_message); 
    commit; 
END; 

BEGIN 
    DBMS_AQADM.ADD_SUBSCRIBER (queue_name => 'message_queue', 
    subscriber => SYS.AQ$_AGENT('queue_subscriber', 'message_queue',NULL)); 
    DBMS_AQ.REGISTER (
    SYS.AQ$_REG_INFO_LIST(
     SYS.AQ$_REG_INFO(
     'MESSAGE_QUEUE:QUEUE_SUBSCRIBER', 
     DBMS_AQ.NAMESPACE_AQ, 
     'plsql://QUEUE_CALLBACK', 
     HEXTORAW('FF') 
    ) 
    ), 1 
); 
END; 

Answer 1

À première vue, il semble que vous ne commençant la file d'attente (dbms_aqadm.start_queue), ne vous enqueueing quoi que ce soit à lui (dbms_aq.enqueue).

Je recommanderais de suivre ce demo.

Answer 2

Vous devez faire attention à la version de la base de données. quelques bugs ont été rapportés sur issues with Oracle Aq. En particulier, j'ai suivi this link pour créer mon propre exemple, en exécutant la démo dans une base de données d'entreprise Oracle 11gR2. J'étais capable de mettre en file d'attente, dequeue, purger la file d'attente mais l'écouteur créé avec Dbms_Aq.Register ne fonctionnait pas. J'ai couru le même exemple en téléchargeant une base de données Oracle 11g R2 xe et cela a fonctionné.

Le même exemple a été exécuté dans une instance Oracle 10gR2 et fonctionne parfaitement.

Il y a certaines choses que vous devez être prudent sur l'utilisation aq:

• utiliser les paramètres appropriés en ajoutant l'abonné
• utiliser l'espace de noms approprié enregistrement de l'auditeur avec Dbms_Aq.Register
• utiliser la indicateur de consommateurs multiples déclarant la table de file d'attente
• utiliser les autorisations appropriées pour les packages et gérer les files d'attente
• utiliser le nom qualifié de la file d'attente s dans certains cas, si cela n'a pas fonctionné.

' d'abord créer le schéma

connect/as sysdba 
-- @?/rdbms/admin/dbmsaqad.sql --(install if you don't have aq installed yet) 

-- create the user and permissions 
create user aqadmin identified by aqadmin default tablespace users temporary tablespace temp; 
GRANT create session TO aqadmin; 
grant connect, resource to aqadmin; 
GRANT aq_administrator_role TO aqadmin IDENTIFIED BY aqadmin; 
GRANT execute ON dbms_aq TO aqadmin; 
GRANT execute ON dbms_aqadm TO aqadmin; 

Créer les objets ddl

CREATE TABLE demo_queue_message_table 
(message VARCHAR2(4000)); 

créer les objets aq-

create or replace type demo_queue_payload_type as object(message varchar2(4000)) ; 
/

begin 
    DBMS_AQADM.CREATE_QUEUE_TABLE (queue_table => 'demo_queue_table', queue_payload_type => 'demo_queue_payload_type',multiple_consumers => TRUE); 
    DBMS_AQADM.CREATE_QUEUE (queue_name => 'demo_queue', queue_table => 'demo_queue_table'); 
    DBMS_AQADM.START_QUEUE('demo_queue'); 
end; 
/

CREATE or replace PROCEDURE demo_queue_callback_procedure(
       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   demo_queue_payload_type; 

BEGIN 

    r_dequeue_options.msgid := descr.msg_id; 
    r_dequeue_options.consumer_name := descr.consumer_name; 

    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 
    ); 

    INSERT INTO demo_queue_message_table (message) 
    VALUES ('Message [' || o_payload.message || '] ' || 
      'dequeued at [' || TO_CHAR(SYSTIMESTAMP, 
             'DD-MON-YYYY HH24:MI:SS.FF3') || ']'); 
    COMMIT; 

END; 
/


BEGIN 

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

    DBMS_AQ.REGISTER (
     SYS.AQ$_REG_INFO_LIST(
      SYS.AQ$_REG_INFO(
      'DEMO_QUEUE:DEMO_QUEUE_SUBSCRIBER', 
      DBMS_AQ.NAMESPACE_AQ, 
      'plsql://DEMO_QUEUE_CALLBACK_PROCEDURE', 
      HEXTORAW('FF') 
      ) 
     ), 
     1 
     ); 
END; 
/

Et enfin tester la file d'attente

DECLARE 

    r_enqueue_options DBMS_AQ.ENQUEUE_OPTIONS_T; 
    r_message_properties DBMS_AQ.MESSAGE_PROPERTIES_T; 
    v_message_handle  RAW(16); 
    o_payload   demo_queue_payload_type; 

BEGIN 

    o_payload := demo_queue_payload_type(
        TO_CHAR(SYSTIMESTAMP, 'DD-MON-YYYY HH24:MI:SS.FF3') 
        ); 

    DBMS_AQ.ENQUEUE(
     queue_name   => 'demo_queue', 
     enqueue_options => r_enqueue_options, 
     message_properties => r_message_properties, 
     payload   => o_payload, 
     msgid    => v_message_handle 
    ); 

    COMMIT; 

END; 
/