Skip to content

Commit b582c8c

Browse files
markxnelsonmarkxnelson
authored
updated the plsql samples (#437)
Signed-off-by: Mark Nelson <[email protected]>
1 parent 567aac0 commit b582c8c

File tree

4 files changed

+123
-235
lines changed

4 files changed

+123
-235
lines changed

‎code-teq/plsqlTeq/cleanupTEQ.sql‎

Lines changed: 15 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -1,15 +1,18 @@
1-
--Clean up all objects related to the obj type: */
2-
Execute DBMS_AQADM.STOP_QUEUE ( queue_name => 'objType_TEQ');
3-
Execute DBMS_AQADM.drop_transactional_event_queue(queue_name =>'objType_TEQ',force=> TRUE);
1+
--
2+
-- This sample demonstrates how to remove (clean up) a TEQ using PL/SQL
3+
--
44

5-
--Cleans up all objects related to the RAW type: */
6-
Execute DBMS_AQADM.STOP_QUEUE ( queue_name => 'rawType_TEQ');
7-
Execute DBMS_AQADM.drop_transactional_event_queue(queue_name =>'rawType_TEQ',force=> TRUE);
5+
-- Execute permission on dbms_aqadm is required.
86

9-
--Cleans up all objects related to the priority queue: */
10-
Execute DBMS_AQADM.STOP_QUEUE ( queue_name => 'jsonType_TEQ');
11-
Execute DBMS_AQADM.drop_transactional_event_queue(queue_name =>'jsonType_TEQ',force=> TRUE);
12-
/
13-
select name, queue_table, dequeue_enabled,enqueue_enabled, sharded, queue_category, recipients from all_queues where OWNER='DBUSER' and QUEUE_TYPE<>'EXCEPTION_QUEUE';
7+
begin
8+
-- first we need to stop the TEQ
9+
dbms_aqadm.stop_queue(
10+
queue_name => 'my_json_teq'
11+
);
12+
13+
-- now we can drop the TEQ
14+
dbms_aqadm.drop_transactional_event_queue(
15+
queue_name => 'my_json_teq'
16+
);
17+
end;
1418
/
15-
EXIT;

‎code-teq/plsqlTeq/createTEQ.sql‎

Lines changed: 44 additions & 101 deletions
Original file line numberDiff line numberDiff line change
@@ -1,107 +1,50 @@
1+
--
2+
-- This sample demonstrates how to create a TEQ using PL/SQL
3+
--
14

2-
CREATE type Message_type as object (subject VARCHAR2(30), text VARCHAR2(80));
3-
/
4-
-- Creating an Object type queue
5-
BEGIN
6-
DBMS_AQADM.CREATE_TRANSACTIONAL_EVENT_QUEUE(
7-
queue_name =>'objType_TEQ',
8-
storage_clause =>null,
9-
multiple_consumers =>true,
10-
max_retries =>10,
11-
comment =>'ObjectType for TEQ',
12-
queue_payload_type =>'Message_type',
13-
queue_properties =>null,
14-
replication_mode =>null);
15-
DBMS_AQADM.START_QUEUE (queue_name=> 'objType_TEQ', enqueue =>TRUE, dequeue=> True);
16-
END;
17-
/
5+
-- There are various payload types supported, including user-defined object, raw, JMS and JSON.
6+
-- This sample uses the JSON payload type.
187

19-
-- Creating a RAW type queue:
20-
BEGIN
21-
DBMS_AQADM.CREATE_TRANSACTIONAL_EVENT_QUEUE(
22-
queue_name =>'rawType_TEQ',
23-
storage_clause =>null,
24-
multiple_consumers =>true,
25-
max_retries =>10,
26-
comment =>'RAW type for TEQ',
27-
queue_payload_type =>'RAW',
28-
queue_properties =>null,
29-
replication_mode =>null);
30-
DBMS_AQADM.START_QUEUE (queue_name=> 'rawType_TEQ', enqueue =>TRUE, dequeue=> True);
31-
END;
32-
/
8+
-- Execute permission on dbms_aqadm is required.
339

34-
--Creating JSON type queue:
35-
BEGIN
36-
DBMS_AQADM.CREATE_TRANSACTIONAL_EVENT_QUEUE(
37-
queue_name =>'jsonType_TEQ',
38-
storage_clause =>null,
39-
multiple_consumers =>true,
40-
max_retries =>10,
41-
comment =>'jsonType for TEQ',
42-
queue_payload_type =>'JSON',
43-
queue_properties =>null,
44-
replication_mode =>null);
45-
DBMS_AQADM.START_QUEUE (queue_name=> 'jsonType_TEQ', enqueue =>TRUE, dequeue=> True);
46-
END;
47-
/
48-
BEGIN
49-
DBMS_AQADM.CREATE_TRANSACTIONAL_EVENT_QUEUE(
50-
queue_name =>'JAVA_TEQ_PUBSUB_QUEUE',
51-
storage_clause =>null,
52-
multiple_consumers=>true,
53-
max_retries =>10,
54-
comment =>'JAVA_TEQ_PUBSUB_QUEUE',
55-
queue_payload_type=>'JMS',
56-
queue_properties =>null,
57-
replication_mode =>null);
58-
DBMS_AQADM.START_QUEUE (queue_name=> 'JAVA_TEQ_PUBSUB_QUEUE', enqueue =>TRUE, dequeue=> True);
59-
END;
10+
begin
11+
-- create the TEQ
12+
dbms_aqadm.create_transactional_event_queue(
13+
-- note, in Oracle 19c this is called create_sharded_queue() but has the same parameters
14+
queue_name => 'my_json_teq',
15+
queue_payload_type => 'JSON',
16+
-- when mutiple_consumers is true, this will create a pub/sub "topic" - the default is false
17+
multiple_consumers => true,
18+
max_retries => 10,
19+
comment => 'A TEQ with JSON payload'
20+
);
21+
22+
-- start the TEQ
23+
dbms_aqadm.start_queue(
24+
queue_name => 'my_json_teq',
25+
-- these two parameters control whether enqueueing and dequeueing will be allowed
26+
enqueue => true,
27+
dequeue => true
28+
);
29+
end;
6030
/
61-
DECLARE
62-
subscriber sys.aq$_agent;
63-
BEGIN
64-
dbms_aqadm.add_subscriber(queue_name => 'objType_TEQ' , subscriber => sys.aq$_agent('teqBasicObjSubscriber' , null ,0), rule => 'correlation = ''teqBasicObjSubscriber''');
6531

66-
dbms_aqadm.add_subscriber(queue_name => 'rawType_TEQ' , subscriber => sys.aq$_agent('teqBasicRawSubscriber' , null ,0), rule => 'correlation = ''teqBasicRawSubscriber''');
32+
--
33+
-- You may also want to create a subscriber for the TEQ, pub/sub topics normally deliver
34+
-- messages only when the consumer/subscriber is present.
35+
--
6736

68-
dbms_aqadm.add_subscriber(queue_name => 'jsonType_TEQ' , subscriber => sys.aq$_agent('teqBasicJsonSubscriber' , null ,0), rule => 'correlation = ''teqBasicJsonSubscriber''');
69-
70-
END;
71-
/
72-
CREATE OR REPLACE FUNCTION enqueueDequeueTEQ(subscriber varchar2, queueName varchar2, message Message_Typ) RETURN Message_Typ
73-
IS
74-
enqueue_options DBMS_AQ.enqueue_options_t;
75-
message_properties DBMS_AQ.message_properties_t;
76-
message_handle RAW(16);
77-
dequeue_options DBMS_AQ.dequeue_options_t;
78-
messageData Message_Typ;
79-
80-
BEGIN
81-
messageData := message;
82-
message_properties.correlation := subscriber;
83-
DBMS_AQ.ENQUEUE(
84-
queue_name => queueName,
85-
enqueue_options => enqueue_options,
86-
message_properties => message_properties,
87-
payload => messageData,
88-
msgid => message_handle);
89-
COMMIT;
90-
DBMS_OUTPUT.PUT_LINE ('----------ENQUEUE Message: ' || 'ORDERID: ' || messageData.ORDERID || ', OTP: ' || messageData.OTP ||', DELIVERY_STATUS: ' || messageData.DELIVERY_STATUS );
91-
92-
dequeue_options.dequeue_mode := DBMS_AQ.REMOVE;
93-
dequeue_options.wait := DBMS_AQ.NO_WAIT;
94-
dequeue_options.navigation := DBMS_AQ.FIRST_MESSAGE;
95-
dequeue_options.consumer_name := subscriber;
96-
DBMS_AQ.DEQUEUE(
97-
queue_name => queueName,
98-
dequeue_options => dequeue_options,
99-
message_properties => message_properties,
100-
payload => messageData,
101-
msgid => message_handle);
102-
COMMIT;
103-
DBMS_OUTPUT.PUT_LINE ('----------DEQUEUE Message: ' || 'ORDERID: ' || messageData.ORDERID || ', OTP: ' || messageData.OTP ||', DELIVERY_STATUS: ' || messageData.DELIVERY_STATUS );
104-
RETURN messageData;
105-
END;
106-
/
107-
EXIT;
37+
declare
38+
subscriber sys.aq$_agent;
39+
begin
40+
dbms_aqadm.add_subscriber(
41+
queue_name => 'my_json_teq',
42+
subscriber => sys.aq$_agent(
43+
'my_subscriber', -- the subscriber name
44+
null, -- address, only used for notifications
45+
0 -- protocol
46+
),
47+
rule => 'correlation = ''my_subscriber'''
48+
);
49+
end;
50+
/

‎code-teq/plsqlTeq/dequeueTEQ.sql‎

Lines changed: 31 additions & 61 deletions
Original file line numberDiff line numberDiff line change
@@ -1,72 +1,42 @@
1-
--Dequeue from obj Type Messages */
2-
DECLARE
3-
dequeue_options dbms_aq.dequeue_options_t;
4-
message_properties dbms_aq.message_properties_t;
5-
message_handle RAW(16);
6-
message Message_type;
7-
8-
BEGIN
9-
dequeue_options.dequeue_mode := DBMS_AQ.REMOVE;
10-
dequeue_options.wait := DBMS_AQ.NO_WAIT;
11-
dequeue_options.navigation := DBMS_AQ.FIRST_MESSAGE;
12-
dequeue_options.consumer_name := 'teqBasicObjSubscriber';
13-
14-
DBMS_AQ.DEQUEUE(
15-
queue_name => 'objType_TEQ',
16-
dequeue_options => dequeue_options,
17-
message_properties => message_properties,
18-
payload => message,
19-
msgid => message_handle);
20-
21-
DBMS_OUTPUT.PUT_LINE ('Message: ' || message.subject || ' ... ' || message.text );
22-
COMMIT;
23-
END;
24-
/
1+
--
2+
-- This sample demonstrates how to enqueue a message onto a TEQ using PL/SQL
3+
--
254

26-
--Dequeue from RAW Type Messages */
27-
DECLARE
28-
dequeue_options DBMS_AQ.dequeue_options_t;
29-
message_properties DBMS_AQ.message_properties_t;
30-
message_handle RAW(16);
31-
message RAW(4096);
32-
33-
BEGIN
34-
dequeue_options.dequeue_mode := DBMS_AQ.REMOVE;
35-
dequeue_options.wait := DBMS_AQ.NO_WAIT;
36-
dequeue_options.navigation := DBMS_AQ.FIRST_MESSAGE;
37-
dequeue_options.consumer_name := 'teqBasicRawSubscriber';
5+
-- There are various payload types supported, including user-defined object, raw, JMS and JSON.
6+
-- This sample uses the JSON payload type.
387

39-
DBMS_AQ.DEQUEUE(
40-
queue_name => 'rawType_TEQ',
41-
dequeue_options => dequeue_options,
42-
message_properties => message_properties,
43-
payload => message,
44-
msgid => message_handle);
45-
COMMIT;
46-
END;
47-
/
8+
-- Execute permission on dbms_aq is required.
489

49-
--Dequeue from JSON TEQ
50-
DECLARE
10+
set serveroutput on;
11+
declare
5112
dequeue_options dbms_aq.dequeue_options_t;
5213
message_properties dbms_aq.message_properties_t;
53-
message_handle RAW(16);
54-
message JSON;
14+
message_handle raw(16);
15+
message json;
5516

56-
BEGIN
57-
dequeue_options.dequeue_mode := DBMS_AQ.REMOVE;
58-
dequeue_options.wait := DBMS_AQ.NO_WAIT;
59-
dequeue_options.navigation := DBMS_AQ.FIRST_MESSAGE;
60-
dequeue_options.consumer_name := 'teqBasicJsonSubscriber';
17+
begin
18+
-- dequeue_mode determines whether we will consume the message or just browse it and leave it there
19+
dequeue_options.dequeue_mode := dbms_aq.remove;
20+
-- wait controls how long to wait for a message to arrive before giving up
21+
dequeue_options.wait := dbms_aq.no_wait;
22+
-- we must specify navigation so we know where to look in the TEQ
23+
dequeue_options.navigation := dbms_aq.first_message;
24+
-- set the consumer name
25+
dequeue_options.consumer_name := 'my_subscriber';
6126

62-
DBMS_AQ.DEQUEUE(
63-
queue_name => 'jsonType_TEQ',
27+
-- perform the dequeue
28+
dbms_aq.dequeue(
29+
queue_name => 'my_json_teq',
6430
dequeue_options => dequeue_options,
6531
message_properties => message_properties,
6632
payload => message,
67-
msgid => message_handle);
68-
dbms_output.put_line(json_serialize(message));
69-
COMMIT;
70-
END;
33+
msgid => message_handle
34+
);
35+
36+
-- print out the message payload
37+
dbms_output.put_line(json_serialize(message));
38+
39+
-- commit the transaction
40+
commit;
41+
end;
7142
/
72-
EXIT;

‎code-teq/plsqlTeq/enqueueTEQ.sql‎

Lines changed: 33 additions & 61 deletions
Original file line numberDiff line numberDiff line change
@@ -1,66 +1,38 @@
1-
--Enqueue to objType Message
2-
DECLARE
3-
enqueue_options dbms_aq.enqueue_options_t;
4-
message_properties dbms_aq.message_properties_t;
5-
message_handle RAW(16);
6-
message Message_type;
1+
--
2+
-- This sample demonstrates how to enqueue a message onto a TEQ using PL/SQL
3+
--
74

8-
BEGIN
9-
message := Message_type('NORMAL MESSAGE','enqueue objType_TEQ');
10-
message_properties.correlation := 'teqBasicObjSubscriber';
5+
-- There are various payload types supported, including user-defined object, raw, JMS and JSON.
6+
-- This sample uses the JSON payload type.
117

12-
DBMS_AQ.ENQUEUE(
13-
queue_name => 'objType_TEQ',
14-
enqueue_options => enqueue_options,
15-
message_properties => message_properties,
16-
payload => message,
17-
msgid => message_handle);
18-
COMMIT;
19-
END;
20-
/
8+
-- Execute permission on dbms_aq is required.
219

22-
--Enqueue to rawType Message
23-
DECLARE
24-
enqueue_options dbms_aq.enqueue_options_t;
25-
message_properties dbms_aq.message_properties_t;
26-
message_handle RAW(16);
27-
message RAW(4096);
28-
29-
BEGIN
30-
message := HEXTORAW(RPAD('FF',4095,'FF'));
31-
message_properties.correlation := 'teqBasicRawSubscriber';
32-
33-
DBMS_AQ.ENQUEUE(
34-
queue_name => 'rawType_TEQ',
35-
enqueue_options => enqueue_options,
36-
message_properties => message_properties,
37-
payload => message,
38-
msgid => message_handle);
39-
COMMIT;
40-
END;
41-
/
42-
-- Enqueue for JSON Message
43-
DECLARE
44-
enqueue_options dbms_aq.enqueue_options_t;
45-
message_properties dbms_aq.message_properties_t;
46-
message_handle RAW(16);
47-
message json;
48-
BEGIN
49-
message:= json('
10+
declare
11+
enqueue_options dbms_aq.enqueue_options_t;
12+
message_properties dbms_aq.message_properties_t;
13+
message_handle raw(16);
14+
message json;
15+
begin
16+
-- create the message payload
17+
message := json('
5018
{
51-
"ORDERID":12345,
52-
"USERNAME":"name"
53-
}');
54-
message_properties.correlation := 'teqBasicJsonSubscriber';
19+
"orderid": 12345,
20+
"username": "Jessica Smith"
21+
}
22+
');
23+
24+
-- set the consumer name
25+
message_properties.correlation := 'my_subscriber';
5526

56-
DBMS_AQ.ENQUEUE(
57-
queue_name => 'jsonType_TEQ',
58-
enqueue_options => enqueue_options,
59-
message_properties => message_properties,
60-
payload => message,
61-
msgid => message_handle);
62-
dbms_output.put_line(json_serialize(message));
63-
COMMIT;
64-
END;
65-
/
66-
EXIT;
27+
-- enqueue the message
28+
dbms_aq.enqueue(
29+
queue_name => 'my_json_teq',
30+
enqueue_options => enqueue_options,
31+
message_properties => message_properties,
32+
payload => message,
33+
msgid => message_handle);
34+
35+
-- commit the transaction
36+
commit;
37+
end;
38+
/

0 commit comments

Comments
 (0)