RIB - JMS AQ

English Version

Neste post partilho algumas queries que usei para monitorizar o processamento de mensagens no RIB.

-- find all items queues
SQL> select * from user_queues where queue_type = 'NORMAL_QUEUE' and name like '%ITEM%'

--- find all items to process
SQL> select * from ETITEMSTLFROMRIBTABLE

-- queue subscribers
SQL> select * from AQ$_ETITEMSTLFROMRIBTABLE_S

-- queue historic
SQL> select * from AQ$_ETITEMSTLFROMRIBTABLE_H where msgid = 'FE4C339218D830B1E0440024813309CD'

--- find all tables/queue tables
SQL> select * from dba_tables where table_name like '%ETITEMSTLFROMRIB%'

Testar Integracao - RMS > WMS

Uma forma facil de testar a integracao (RIB) apos adicionar um novo warehouse (WMS), e actualizar um Item no RMS e validar se as alteracoes estao a ser transmitidas ao WMS, no exemplo seguinte mostro como fazer isso, e os tabelas/queues envolvidas.

Item : 101715654

-- RMS DB
SQL> select  item, item_desc  from item_master where item ='101715654'

ITEM ITEM_DESC
101715654 ITEM XPTO

SQL> update  rms.item_master
set item_desc = 'ITEM XPTO - DESC CHANGED'
where item='101715654';

SQL> select  item, item_desc  from item_master where item ='101715654';
SQL> commit;

-- RIB DB
-- if you are fast enough or if you have retention on the queue 

SQL> SELECT msgid,enq_time, deq_time, r.user_data.text_lob
FROM RIBAQ.ETITEMSFROMRMSTABLE r
WHERE
enq_time > sysdate - 1
and r.user_data.text_lob  like '%101715654%'
ORDER BY enq_time;

--WMS DB 
SQL> select  item_id, description  from wms.item_master where item_id ='101715654';
101715654 ITEM XPTO - DESC CHANGED

Vendor

-- find user queues
SQL> select * from user_queues where queue_type = 'NORMAL_QUEUE' and name like '%VENDOR%'

SQL> select * from user_queues
where
name like '%VEND%' and QUEUE_TYPE = 'NORMAL_QUEUE'
and name not like '%WH%'

--count number vendors
SQL> select count(0) from ETVENDORFROMRMSTABLE

-- select all
SQL> select * from ETVENDORFROMRMSTABLE

-- Validate who subscribed the the message (Query 1)
SQL> select subscriber#,count(0)  from AQ$_ETVENDORFROMRMSTABLE_H
where dequeue_time is null
group by subscriber#

-- Validate who subscribed the the message (Query 2)
SQL> select * from  AQ$_ETVENDORFROMRMSTABLE_S  where subscriber_id in (
select subscriber#  from AQ$_ETVENDORFROMRMSTABLE_H group by subscriber#)

Orders

--Order queues
SQL> select * from ETSTOCKORDERSISOTABLE

--Count orders to WH3
SQL> select count(*) from ETSTKORDERSFROMRIBTOWH3T;

-- WMS Queues
SQL> select * from user_tables 
where table_name like '%ORDERS%' and table_name not like '%WH%'

ASNOut

SQL> alter session set nls_date_format = 'dd-mm-yy hh24:mi'

SQL> select * from user_queues where name like '%ASN%' and QUEUE_TYPE = 'NORMAL_QUEUE' and name not like '%WH%'

--tafr
SQL> select min(enq_time),sysdate as current_time from ETASNOUTTABLE,dual
select count(0), sysdate from ETASNOUTTABLE,dual

-- rms
SQL> select min(enq_time),sysdate as current_time from ETASNOUTATTABLE,dual

SQL> select count(0),sysdate as current_time from ETASNOUTATTABLE,dual

-- look into payloads
SQL> select count(0) from
(SELECT enq_time, deq_time, r.user_data.text_lob
FROM ribaq.ETASNOUTATTABLE r
WHERE
 r.user_data.text_lob like '118000001000%')

SQL> SELECT enq_time, deq_time, r.user_data.text_lob
FROM ribaq.ETASNOUTATTABLE r
WHERE
 r.user_data.text_lob like '118000001000%'

Neste post mostro como encontrar  queue tables ( inventory adjustments neste caso) e o estado das mensagens.

SQL> select owner,table_name from all_tables where table_name like 'ET%INV%';

OWNER                          TABLE_NAME
------------------------------ ------------------------------
RIBAQ                          ETINVADJUSTTABLE
RIBAQ                          ETINVREQTABLE

SQL> select count(0) from RIBAQ.ETINVADJUSTTABLE;

  COUNT(0)
----------
       544

SQL> select msgid,enq_time,deq_time  from  RIBAQ.ETINVADJUSTTABLE order by 2 asc

MSGID                                                             ENQ_TIME                                                                    DEQ_TIME
--------------------------------                                  --------------------------------------------------------------------------- ---------
6511FD971CEBC0AEE0530D2810AC2B8D   13-FEB-18 10.54.58.585547 PM
6511F6994E6C6F6AE0530D2810AC0D9A     13-FEB-18 10.54.58.557901 PM
-- trunctate output

549 rows selected.

SQL> set lines 200
select dq.name,
--dq.queue_type, 
qi.*
from v$aq qi,  dba_queues dq where qi.qid = dq.qid and dq.queue_table like 'ETINVADJUSTTABLE';

NAME                                           QID         WAITING      READY    EXPIRED AVERAGE_MSG_AGE     TOTAL_WAIT AVERAGE_WAIT
------------------------------               ----------   ----------           ----------      ----------       ---------------                   ----------             ------------
ETINVADJUST                              88311           0                 549          0                 65073                        35725543          65073.8488
AQ$_ETINVADJUSTTABLE_E     88310           0                 0              0                 0                                0                         0

Como extratir mensagens/payloads Oracle AQ/JMS queues/topicos ?

SQL> set lines 300
SQL> col enq_time  form a30
SQL> col deq_time form a10
SQL> col msg form a100
SQL> set longchunksize 30000
SQL> set long 30000

SQL> select msgid,enq_time, deq_time, r.user_data.text_lob msg
 from ribaq.etitemsfromrmstable r
 where
 enq_time > sysdate - 1
 and r.user_data.text_lob  like '%101715654%'
 order by enq_time;

Como alterar propriedades das queues ?
Exemplo max_retries,retention

SQL> select qid,owner,name,queue_table,queue_type,max_retries,retry_delay,retention, enqueue_enabled,dequeue_enabled 
from dba_queues where (name like '%ETRECEIVING%' or name like '%ASNOUTAT%')
and queue_type = 'NORMAL_QUEUE' ; 

318421 RIBAQ ETRECEIVING ETRECEIVINGTABLE NORMAL_QUEUE 5 0 0   YES     YES  
311104 RIBAQ ETASNOUTAT ETASNOUTATTABLE   NORMAL_QUEUE 5 0 0   YES     YES  

-- max_retries 200
SQL> EXECUTE dbms_aqadm.alter_queue ( queue_name => 'ETRECEIVING', max_retries => 200 ); 
SQL> EXECUTE dbms_aqadm.alter_queue ( queue_name => 'ETASNOUTAT', max_retries => 200 ); 

--retention 1 day 
SQL> EXECUTE dbms_aqadm.alter_queue ( queue_name => 'ETRECEIVING', retention_time     => 86400 ); 
EXECUTE dbms_aqadm.alter_queue ( queue_name => 'ETASNOUTAT', retention_time     => 86400 ); 

SQL> select qid,owner,name,queue_table,queue_type,max_retries,retry_delay,retention, enqueue_enabled,dequeue_enabled 
from dba_queues where (name like '%ETRECEIVING%' or name like '%ASNOUTAT%')
and queue_type = 'NORMAL_QUEUE' ; 

  318421 RIBAQ ETRECEIVING ETRECEIVINGTABLE NORMAL_QUEUE 200 0 86400   YES     YES  
  311104 RIBAQ ETASNOUTAT ETASNOUTATTABLE   NORMAL_QUEUE 200 0 86400   YES     YES   

--- bulk situations - all queues in the schema 

SQL> select  * from all_queues where queue_type = 'NORMAL_QUEUE' and owner like 'RIBAQ' and retention <> '604800'

SQL> declare

stm_text varchar(200);
cursor c1 is (  select  name from all_queues where queue_type = 'NORMAL_QUEUE' and owner like 'RIBAQ' );
   
begin
 for i in c1  
  loop
   stm_text := 'begin dbms_aqadm.alter_queue ( queue_name => '''|| i.name  ||''', retention_time => 604800); end;';
   dbms_output.put_line(stm_text); 
   EXECUTE IMMEDIATE stm_text;
   
  end loop;
end;
/

Mover mensagem da ERROR queue para Normal Queue

-- RIB QUEUES
-- check messages queues
SQL> select * from ETPRMPRCCHGFROMRPMTABLE  where deq_time is null
SQL> select count(0) from AQ$_ETPRMPRCCHGFROMRPMTABLE

SQL> BEGIN
dbms_aqadm.start_queue('AQ$_ETPRMPRCCHGFROMRPMTABLE_E',enqueue=>FALSE, dequeue=>TRUE);
END;
/

--
-- run it  until the  error queue have none records
--

SQL> DECLARE
  enqueue_options DBMS_AQ.enqueue_options_t;
  dequeue_options dbms_aq.dequeue_options_t;
  message_properties DBMS_AQ.message_properties_t;
  message_handle RAW(16);
  MESSAGE SYS.AQ$_JMS_MESSAGE;
BEGIN
  DBMS_AQ.DEQUEUE( queue_name => 'AQ$_ETPRMPRCCHGFROMRPMTABLE_E', dequeue_options => dequeue_options, message_properties => message_properties, payload => MESSAGE, msgid => message_handle);
  DBMS_AQ.ENQUEUE( queue_name => 'ETPRMPRCCHGFROMRPM', enqueue_options => enqueue_options, message_properties => message_properties, payload => MESSAGE, msgid => message_handle);
END;
/

SQL> commit;

SQL> BEGIN
dbms_aqadm.stop_queue('AQ$_ETPRMPRCCHGFROMRPMTABLE_E');
END;
/

--
-- Other JMS QUEUES - change messages from error queue into the normal
--

-- check messages queues
SQL> select * from NB_JMS_COSTCHG_TEXT where deq_time is null

SQL> BEGIN
dbms_aqadm.start_queue('RMSPRD.AQ$_NB_JMS_COSTCHG_TEXT_E',enqueue=>FALSE, dequeue=>TRUE);
END;
/

--
-- run it  until the  error queue have none records
--

SQL> DECLARE
  enqueue_options DBMS_AQ.enqueue_options_t;
  dequeue_options dbms_aq.dequeue_options_t;
  message_properties DBMS_AQ.message_properties_t;
  message_handle RAW(16);
  MESSAGE SYS.AQ$_JMS_TEXT_MESSAGE;
BEGIN
  DBMS_AQ.DEQUEUE( queue_name => 'RMSPRD.AQ$_NB_JMS_COSTCHG_TEXT_E', dequeue_options => dequeue_options, message_properties => message_properties, payload => MESSAGE, msgid => message_handle);
  DBMS_AQ.ENQUEUE( queue_name => 'RMSPRD.NB_JMS_COSTCHG_QUEUE', enqueue_options => enqueue_options, message_properties => message_properties, payload => MESSAGE, msgid => message_handle);
END;
/

SQL> commit;

Mais info sobre este tema :
https://docs.oracle.com/cd/E82085_01/141/rib_integration_guide/Index.htm

Obrigado pela leitura,
Espero que este post o tenha ajudado.

Rogerio