Oracle AQ Setup

Submitted by code_admin on Mon, 07/23/2018 - 14:52

For AQ Notes

Clean and refresh Queue Subscription.
This removes a single subscriber and all it's registrations from a queue and resets it up.

  1. PROMPT "Start of Create_BPEL_Q_Subscriptions.sql"
  2. set serveroutput on;
  3. /
  4. declare
  5.   l_subscriberName varchar2(1024) := 'ICSTUDENT';
  6.   l_rule varchar2(4096) := 'tab.user_data.event_name IN (''ic.oracle.apps.ar.hz.PersonBO.create'',''ic.oracle.apps.ar.hz.PersonBO.update'',''ic.oracle.apps.ar.hz.PersonCustBO.update'',''ic.oracle.apps.ar.hz.PersonContBO.sync'',''ic.oracle.apps.per.personTypeUsage.Sync'')';
  7.   l_packagename                 VARCHAR2(4096) := 'APPS.ICSOA_STUDENT_PKG';
  8.   l_procedurename               VARCHAR2(4096) := 'EventSubscriber';
  9.  
  10.  
  11.     l_subscriber       sys.aq$_agent;
  12.     l_queueNameSHORT varchar2(1024) := 'WF_BPEL_Q';
  13.     l_queueName varchar2(1024) := 'APPS.' || l_queueNameSHORT;
  14.  
  15.   l_reginfo             sys.aq$_reg_info;
  16.   l_reg_list            sys.aq$_reg_info_list;
  17.  
  18.  
  19.   cursor c_subscriptions(p_consumer_name in varchar2,p_queue_name in varchar2) is
  20.     select s.*
  21.     from "SYS"."DBA_QUEUE_SUBSCRIBERS" s
  22.     where s.queue_name=p_queue_name
  23.     and s.consumer_name=p_consumer_name
  24.   ;
  25.   TYPE t_subscriptions IS TABLE OF c_subscriptions%ROWTYPE INDEX BY PLS_INTEGER;  
  26.   l_subscriptions t_subscriptions;
  27.   l_cur number;  
  28.  
  29.   cursor c_regs(p_consumer_name in varchar2,p_queue_name in varchar2) is
  30.     select r.*
  31.     from "SYS".DBA_SUBSCR_REGISTRATIONS r
  32.     where r.subscription_name like '%' || p_queue_name || '%:%' || p_consumer_name || '%'
  33.   ;
  34.   TYPE t_regs IS TABLE OF c_regs%ROWTYPE INDEX BY PLS_INTEGER;  
  35.   l_regs t_regs;
  36.   l_curReg number;  
  37.  
  38.   l_num number;
  39. begin
  40.   DBMS_OUTPUT.ENABLE(1000000);
  41.   dbms_output.put_line('START Create_BPEL_Q_Subscriptions.sql');
  42.  
  43.   open c_subscriptions(l_subscriberName, l_queueNameSHORT);
  44.   fetch c_subscriptions bulk collect into l_subscriptions;
  45.   close c_subscriptions;
  46.  
  47.   l_cur := l_subscriptions.FIRST;
  48.   while (l_cur is not null) LOOP
  49.     dbms_output.put_line('Deleting Subscription ' || l_subscriptions(l_cur).consumer_name);
  50.      l_subscriber := sys.aq$_agent(l_subscriptions(l_cur).consumer_name,NULL, NULL);
  51.      DBMS_AQADM.REMOVE_SUBSCRIBER(
  52.         queue_name => l_queueName,
  53.         subscriber => l_subscriber
  54.       );
  55.      
  56.         l_cur := l_subscriptions.NEXT(l_cur);
  57.   END LOOP; --Loop through subs  
  58.  
  59.   dbms_output.put_line('Deleting Registrations');
  60.   open c_regs(l_subscriberName, l_queueNameSHORT);
  61.   fetch c_regs bulk collect into l_regs;
  62.   close c_regs;
  63.   l_curReg := l_regs.FIRST;
  64.   while (l_curReg is not null) LOOP
  65.         dbms_output.put_line('Deleting Registration - ' || l_regs(l_curReg).subscription_name || '-' || l_regs(l_curReg).location_name);
  66.         l_reginfo := sys.aq$_reg_info(
  67.                             l_regs(l_curReg).subscription_name,
  68.                             DBMS_AQ.NAMESPACE_AQ,
  69.                             l_regs(l_curReg).location_name,
  70.                             l_regs(l_curReg).user_context
  71.                       );
  72.         l_reg_list  := sys.aq$_reg_info_list(l_reginfo);  
  73.         dbms_aq.unregister(l_reg_list,1);  
  74.        
  75.         l_curReg := l_regs.NEXT(l_curReg);
  76.   END LOOP; --Loop through regs  
  77.    
  78.   --Check there are no registrations
  79.   select count('x')
  80.   into l_num
  81.   from "SYS".DBA_SUBSCR_REGISTRATIONS
  82.   where subscription_name like '%' || l_queueNameSHORT || '%:%' || l_subscriberName || '%'  
  83.   ;
  84.   if l_num<>0 then
  85.     raise_application_error(-20000,'ERROR Registraiton exists (should have been deleted):' || to_char(l_num));
  86.   end if;
  87.  
  88.   --check there are no subscribers
  89.   select count('x')
  90.   into l_num
  91.   from "SYS"."DBA_QUEUE_SUBSCRIBERS"
  92.   where queue_name=l_queueNameSHORT
  93.   and consumer_name=l_subscriberName  
  94.   ;
  95.   if l_num<>0 then
  96.     raise_application_error(-20000,'ERROR Subscription exists (should have been deleted):' || to_char(l_num));
  97.   end if;
  98.  
  99.  
  100.     dbms_output.put_line('Adding Subscription ' || l_subscriberName);
  101.     dbms_output.put_line('  for rule ' || l_rule);
  102.    l_subscriber := sys.aq$_agent(l_subscriberName,NULL, NULL);
  103.    DBMS_AQADM.ADD_SUBSCRIBER (
  104.       queue_name => l_queueName,
  105.       subscriber => l_subscriber,
  106.       rule => l_rule
  107.     );
  108.   l_reginfo := sys.aq$_reg_info(
  109.                       l_queueName || ':' || l_subscriberName,
  110.                       DBMS_AQ.NAMESPACE_AQ,
  111.                             'plsql://' || l_packagename || '.' || l_procedurename || '?PR=0',
  112.                       HEXTORAW('FF')
  113.                 );
  114.   l_reg_list  := sys.aq$_reg_info_list(l_reginfo);  
  115.     dbms_output.put_line('Adding Registration plsql://' || l_packagename || '.' || l_procedurename || '?PR=0');
  116.   dbms_aq.register(l_reg_list,1);  
  117.  
  118.  
  119.   --Check there is 1 registrations
  120.   select count('x')
  121.   into l_num
  122.   from "SYS".DBA_SUBSCR_REGISTRATIONS
  123.   where subscription_name like '%' || l_queueNameSHORT || '%:%' || l_subscriberName || '%'  
  124.   ;
  125.   if l_num<>1 then
  126.     raise_application_error(-20000,'ERROR Registraiton not created:' || to_char(l_num));
  127.   end if;
  128.  
  129.   --check there is 1 subscribers
  130.   select count('x')
  131.   into l_num
  132.   from "SYS"."DBA_QUEUE_SUBSCRIBERS"
  133.   where queue_name=l_queueNameSHORT
  134.   and consumer_name=l_subscriberName  
  135.   ;
  136.   if l_num<>1 then
  137.     raise_application_error(-20000,'ERROR Subscription not created:' || to_char(l_num));
  138.   end if;  
  139.  
  140.   dbms_output.put_line('END Create_BPEL_Q_Subscriptions.sql');
  141.   commit;
  142. end;
  143. /
  144. PROMPT "End of Create_BPEL_Q_Subscriptions.sql"
RJM Article Type
Quick Reference