Results 1 to 8 of 8
  1. #1
    Join Date
    May 2012
    Posts
    4

    Unanswered: Advanced Queueing - Callback routine doesn't execute

    Running 11g Enterprise - Rel 11.2.0.2.0. I've created a queue, corresponding queue table, added a subscriber, and registered a PL/SQL Callback routine. All enqueuing/dequeuing is occurring within the context of a single user.

    I can enqueue messages and see them in the queue table. I was able to confirm that the callback routine itself wasn't failing. Either the callback routine is never getting called, or the internal call to it is failing. What am I missing? Any tips on debugging via log/trace files?

    Code below....Thanks in advance!


    Code:
    -- Step:1 Create Queue Table
    begin
        dbms_aqadm.create_queue_table
            (
            queue_table => 'CHImporterQ_TABLE',
            queue_payload_type => 'CHD.BATCHSUBMESSAGETYPE',
            sort_list => 'PRIORITY, ENQ_TIME',
            multiple_consumers => true,
            compatible => '10.0.0',
            primary_instance => 0,
            secondary_instance => 0,
            comment => 'CH Batch File CHImporterQ queue',
            storage_clause => 'tablespace RDATA pctfree 10 initrans 1 maxtrans 255 storage ( initial 64K minextents 1 maxextents unlimited )'
            );
    end;
    
    -- Step:2 Create Queue
    begin
        dbms_aqadm.create_queue
            (
            queue_name => 'CHD.CHImporterQ',
            queue_table => 'CHD.CHImporterQ_TABLE',
            queue_type => sys.dbms_aqadm.NORMAL_QUEUE,
            max_retries => 5,
            retry_delay => 0,
            retention_time => 0
            );
    end;
    
    -- Step:3 Start Queue
    begin
        dbms_aqadm.start_queue(queue_name=>'CHD.CHImporterQ',enqueue => TRUE, dequeue => TRUE);
    end;
    
    -- Step:4  Add Subscriber
    begin
        dbms_aqadm.add_subscriber
            (
            QUEUE_NAME => 'CHD.CHImporterQ',
            subscriber => SYS.AQ$_AGENT(
                                  'CHImporterQ_SUBSCRIBER', 
                                  NULL, 
                                  NULL)
            );
    end;
    
    -- Step:5 Register callback procedure
    begin
        DBMS_AQ.REGISTER 
            (
            SYS.AQ$_REG_INFO_LIST(
                SYS.AQ$_REG_INFO
                    (
                    'CHD.CHImporterQ:CHImporterQ_SUBSCRIBER',
                    DBMS_AQ.NAMESPACE_AQ, 'PLSQL://CHD.BATCH_SUBMISSIONS.CHIMPORTERQ_CALLBACK',
                    HEXTORAW('FF')
                    )
                ),
              1
            );
              
       END;
    END;


    Code for enqueuing message is as follows:

    Code:
    DECLARE
        r_enqueue_options    dbms_aq.enqueue_options_t;
        r_message_properties dbms_aq.message_properties_t;
        v_message_handle     RAW(16);
        o_payload                 batchsubmessagetype;
    BEGIN
        o_payload := batchsubmessagetype
                               (
                               var_numnextdocid, 
                               var_BatchTypeID,
                               var_ReferenceId
                               );
    
        dbms_aq.enqueue
            (
            queue_name         => 'CHIMPORTERQ',
            enqueue_options    => r_enqueue_options,
            message_properties => r_message_properties,
            payload            => o_payload,
            msgid              => v_message_handle
            );
    END;
    The signature for the callback procedure is as follows:

    Code:
    PROCEDURE CHImporterQ_Callback 
        (
        CONTEXT  RAW,
        reginfo  SYS.AQ$_REG_INFO,
        descr    SYS.AQ$_DESCRIPTOR,
        payload  RAW,
        payloadl NUMBER
        ) AS ...
    Last edited by jm21; 05-11-12 at 16:20.

  2. #2
    Join Date
    Aug 2003
    Location
    Where the Surf Meets the Turf @Del Mar, CA
    Posts
    7,776
    Provided Answers: 1
    post COPY & PASTE of whole session showing actual execution of the posted code
    You can lead some folks to knowledge, but you can not make them think.
    The average person thinks he's above average!
    For most folks, they don't know, what they don't know.
    Good judgement comes from experience. Experience comes from bad judgement.

  3. #3
    Join Date
    May 2012
    Posts
    4
    Executed code below. Thanks for looking at this.

    Code:
     
    SQL> 
    SQL> -- Create Test Table (Callback procedure attempts to insert record here).
    SQL> CREATE TABLE q_test
      2      (
      3      test_msg    VARCHAR2(100)
      4      );
     
    Table created
    SQL> -- Create payload object type.
    SQL> CREATE TYPE TESTBATCHTYPE AS OBJECT
      2      (
      3      BatchSub_Id       NUMBER,
      4      Batchsub_Type_Id  Number,
      5      Entity_Reference_Id  Number
      6      );
      7  /
     
    Type created
    SQL> -- Create Queue Table
    SQL> begin
      2      dbms_aqadm.create_queue_table
      3          (
      4          queue_table => 'CHImporterQ_TABLE',
      5          queue_payload_type => 'TESTBATCHTYPE',
      6          sort_list => 'PRIORITY, ENQ_TIME',
      7          multiple_consumers => true,
      8          compatible => '10.0.0',
      9          primary_instance => 0,
     10          secondary_instance => 0,
     11          comment => 'CH Batch File CHImporterQ queue',
     12          storage_clause => 'tablespace RDATA pctfree 10 initrans 1 maxtrans 255 storage ( initial 64K minextents 1 maxextents unlimited )'
     13          );
     14  end;
     15  /
     PL/SQL procedure successfully completed
    
    
    SQL> -- Create Queue
    SQL> begin
      2      dbms_aqadm.create_queue
      3          (
      4          queue_name => 'CHImporterQ',
      5          queue_table => 'CHImporterQ_TABLE',
      6          queue_type => sys.dbms_aqadm.NORMAL_QUEUE,
      7          max_retries => 5,
      8          retry_delay => 0,
      9          retention_time => 0
     10          );
     11  end;
     12  /
     PL/SQL procedure successfully completed
    
    
    SQL> -- Start Queue
    SQL> begin
      2      dbms_aqadm.start_queue (queue_name=>'CHImporterQ',enqueue => TRUE,dequeue => TRUE);
      3  end;
      4  /
     PL/SQL procedure successfully completed
    
    
    SQL> -- Add Subscriber
    SQL> begin
      2      dbms_aqadm.add_subscriber
      3          (
      4          QUEUE_NAME => 'CHImporterQ',
      5          subscriber => SYS.AQ$_AGENT(
      6                                'CHImporterQ_SUBSCRIBER',
      7                                NULL,
      8                                NULL)
      9          );
     10  end;
     11  /
     PL/SQL procedure successfully completed
    
    
    SQL> -- Create Callback Procdure;
    SQL> CREATE OR REPLACE PROCEDURE CHImporterQ_CallbackTest
      2      (
      3      CONTEXT     IN RAW,
      4      reginfo     IN SYS.aq$_reg_info,
      5      descr       IN SYS.aq$_descriptor,
      6      payload     IN RAW,
      7      payloadl    IN NUMBER
      8      ) AS
      9  -- Local Vars.
     10  r_dequeue_options    DBMS_AQ.DEQUEUE_OPTIONS_T;
     11  r_message_properties DBMS_AQ.MESSAGE_PROPERTIES_T;
     12  v_message_handle     RAW(16);
     13  v_payload            TESTBATCHTYPE;
     14  
     15  BEGIN
     16  
     17      r_dequeue_options.msgid := descr.msg_id;
     18      r_dequeue_options.consumer_name := 'CHIMPORTERQ_SUBSCRIBER';
     19  
     20      DBMS_AQ.DEQUEUE
     21          (
     22          queue_name         => descr.queue_name,
     23          dequeue_options    => r_dequeue_options,
     24          message_properties => r_message_properties,
     25          payload            => v_payload,
     26          msgid              => v_message_handle
     27          );
     28  
     29      INSERT INTO
     30          q_test
     31      VALUES
     32          ('BatchSubID: ' || v_payload.BatchSub_Id || ', Batchsub_Type_Id: ' || v_payload.BatchSub_Type_Id);
     33  
     34      COMMIT;
     35  END CHImporterQ_CallbackTest;
     36  /
    Procedure created
    
    
    SQL> -- Register callback procedure
    SQL> begin
      2      DBMS_AQ.REGISTER
      3          (
      4          SYS.AQ$_REG_INFO_LIST(
      5              SYS.AQ$_REG_INFO
      6                  (
      7                  'CHImporterQ:CHImporterQ_SUBSCRIBER',
      8                  DBMS_AQ.NAMESPACE_AQ, 'plsql://CHImporterQ_CallbackTest',
      9                  HEXTORAW('FF')
     10                  )
     11              ),
     12            1
     13          );
     14  end;
     15  /
    PL/SQL procedure successfully completed
    
    
    SQL> -- Test it.
    SQL> DECLARE
      2      r_enqueue_options    dbms_aq.enqueue_options_t;
      3      r_message_properties dbms_aq.message_properties_t;
      4      v_message_handle     RAW(16);
      5      v_payload            TESTBATCHTYPE;
      6  BEGIN
      7      -- Create payload object
      8      v_payload := TESTBATCHTYPE(1,1,10);
      9  
     10      -- Add to Queue.
     11      dbms_aq.enqueue
     12          (
     13          queue_name         => 'CHIMPORTERQ',
     14          enqueue_options    => r_enqueue_options,
     15          message_properties => r_message_properties,
     16          payload            => v_payload,
     17          msgid              => v_message_handle
     18          );
     19  END;
     20  /
    PL/SQL procedure successfully completed
    
    
    SQL> select * from chimporterq_table; -- I see the message in the queue table.
     
    Q_NAME                         MSGID                            CORRID                                                                             PRIORITY      STATE DELAY                                             EXPIRATION TIME_MANAGER_INFO                                 LOCAL_ORDER_NO   CHAIN_NO       CSCN       DSCN ENQ_TIME                                          ENQ_UID                        ENQ_TID                        DEQ_TIME                                          DEQ_UID                        DEQ_TID                        RETRY_COUNT EXCEPTION_QSCHEMA              EXCEPTION_QUEUE                   STEP_NO RECIPIENT_KEY DEQUEUE_MSGID                    SENDER_NAME                    SENDER_ADDRESS                                                                   SENDER_PROTOCOL USER_DATA USER_PROP
    ------------------------------ -------------------------------- -------------------------------------------------------------------------------- ---------- ---------- ------------------------------------------------- ---------- ------------------------------------------------- -------------- ---------- ---------- ---------- ------------------------------------------------- ------------------------------ ------------------------------ ------------------------------------------------- ------------------------------ ------------------------------ ----------- ------------------------------ ------------------------------ ---------- ------------- -------------------------------- ------------------------------ -------------------------------------------------------------------------------- --------------- --------- ---------
    CHIMPORTERQ                    BFC86291425ACCE0E040A8C0040169E9                                                                                           1          0                                                                                                                             0          0          0          0 11-MAY-12 07.14.25.895737 PM                      CHDEV                          8.27.86367                                                                                                                                               0                                                                        0             0                                                                                                                                                                  <Object>  <Object>
    
    
    
    SQL> select * from q_test; -- The callback should have inserted a record into this table.
     
    TEST_MSG
    --------------------------------------------------------------------------------
     
    SQL>

  4. #4
    Join Date
    Aug 2003
    Location
    Where the Surf Meets the Turf @Del Mar, CA
    Posts
    7,776
    Provided Answers: 1
    What would you expect to occur if at this point in time you did as below?

    ROLLBACK;

    Who/what can see uncommitted DML?
    You can lead some folks to knowledge, but you can not make them think.
    The average person thinks he's above average!
    For most folks, they don't know, what they don't know.
    Good judgement comes from experience. Experience comes from bad judgement.

  5. #5
    Join Date
    May 2012
    Posts
    4
    I don't believe a ROLLBACK would make any difference? Are you referring to the record that the callback attempted to insert into the Q_TEST table? There is a COMMIT after the insert, and it shouldn't impact the ability to see the record returned in the final SELECT as it's all occurring in the same session.

    Maybe I'm missing your point?

  6. #6
    Join Date
    Oct 2002
    Location
    Cape Town, South Africa
    Posts
    253
    I feel your pain. I also had trouble getting my callback to fire off too.

    Something I did notice though is that in your example session above, you have not commit since you enqueue the message.

    Oracle AQ runs within your active transaction, so you have to commit after enqueue. Only once commited, then the message is available for the aq background job.

    The aq job runs in background, so it won't execute immediately as you commit. It may take some time and is dependent on your job_queue_processes parameter.

    Here's my setting:
    Code:
    dayneo@RMSD> show parameters job
    
    NAME                                 TYPE        VALUE
    ------------------------------------ ----------- ------------------------------
    job_queue_processes                  integer     10
    dayneo@RMSD>

  7. #7
    Join Date
    May 2012
    Posts
    4
    Quote Originally Posted by dayneo View Post
    I feel your pain. I also had trouble getting my callback to fire off too.

    Something I did notice though is that in your example session above, you have not commit since you enqueue the message.

    Oracle AQ runs within your active transaction, so you have to commit after enqueue. Only once commited, then the message is available for the aq background job.

    The aq job runs in background, so it won't execute immediately as you commit. It may take some time and is dependent on your job_queue_processes parameter.

    Here's my setting:
    Code:
    dayneo@RMSD> show parameters job
    
    NAME                                 TYPE        VALUE
    ------------------------------------ ----------- ------------------------------
    job_queue_processes                  integer     10
    dayneo@RMSD>
    Yes. This was preventing my simplified example from working (because it all runs in the same session). The "real" code was suffering from a different issue. After you pointed this out, it gave me confidence that queueing was working correctly on the server and it wasn't a low-level Oracle configuration issue. So I started looking elsewhere and was able to find the problem. Thanks.

  8. #8
    Join Date
    Jun 2013
    Posts
    1

    Same problem

    Could you please explain the solution you found, I have the same problem...

    Thanks

Posting Permissions

  • You may not post new threads
  • You may not post replies
  • You may not post attachments
  • You may not edit your posts
  •