2) GRANT EXECUTE ON SYS.DBMS_AQADM TO BGUMUS;
GRANT EXECUTE ON SYS.DBMS_AQ TO BGUMUS;
2)CREATE TYPE message_t AS OBJECT (json VARCHAR2(4000));
3) BEGIN
DBMS_AQADM.CREATE_QUEUE_TABLE(
queue_table => 'myqueue_tab',
queue_payload_type => 'message_t'
);
-- Create a queue using the new queue table.
DBMS_AQADM.CREATE_QUEUE(
queue_name => 'myqueue',
queue_table => 'myqueue_tab'
);
DBMS_AQADM.START_QUEUE(
queue_name => 'myqueue'
);
END;
4) CREATE PROCEDURE enqueue_message(payload VARCHAR2) AS
msg message_t := message_t(NULL);
msg_id RAW(16);
priority NUMBER;
enqueue_options DBMS_AQ.ENQUEUE_OPTIONS_T;
message_properties DBMS_AQ.MESSAGE_PROPERTIES_T;
BEGIN
msg.json := payload;
message_properties.priority := 1; -- give all messages same priority
DBMS_AQ.ENQUEUE(
queue_name => 'myqueue',
enqueue_options => enqueue_options,
message_properties => message_properties,
payload => msg,
msgid => msg_id);
END;
5) CREATE PROCEDURE dequeue_message(payload OUT VARCHAR2) AS
msg message_t := message_t(NULL);
msg_id RAW(16);
dequeue_options DBMS_AQ.DEQUEUE_OPTIONS_T;
message_properties DBMS_AQ.MESSAGE_PROPERTIES_T;
BEGIN
DBMS_AQ.DEQUEUE(
queue_name => 'myqueue',
dequeue_options => dequeue_options,
message_properties => message_properties,
payload => msg,
msgid => msg_id
);
payload := msg.json;
END;
7) CREATE TABLE customer (
customer_id INTEGER NOT NULL PRIMARY KEY,
name VARCHAR2(40),
address VARCHAR2(40),
city VARCHAR2(40)
);
CREATE TRIGGER customer_queue_trig
AFTER INSERT OR UPDATE OR DELETE ON customer
FOR EACH ROW
DECLARE
json VARCHAR2(4000);
BEGIN
-- For simplicity sake I am just doing simple concatenations. Production
-- JSON serialization code should do character escaping (double quotes,
-- newlines, etc).
IF INSERTING OR UPDATING THEN
json := '{"id":' || :new.customer_id || ',"name":"' || :new.name || '"';
json := json||',"address":"' || :new.address || '","city":"';
json := json|| :new.city || '","dml_type":"';
json := json|| CASE WHEN INSERTING THEN 'I' ELSE 'U' END || '"}';
ELSE
json := '{"id":' || :old.customer_id || ',"dml_type":"D"}';
END IF;
enqueue_message(json);
END;
8)DECLARE
payload VARCHAR2(4000);
BEGIN
dequeue_message(payload);
dbms_output.put_line(payload);
END;
//Oracle.DataAccess.dll 2.111.7.20
// System.Data.OracleClient 4.0.0.0
9)
class OraTest
{
OracleConnection con;
public void Connect()
{
con = new OracleConnection();
con.ConnectionString = @"Data Source = (DESCRIPTION =
(ADDRESS_LIST = (ADDRESS = (PROTOCOL = TCP)(HOST = DESKTOP-U223FLP)(PORT = 1521)))
(CONNECT_DATA = (SERVER = DEDICATED)(SERVICE_NAME = XE)));
User Id = taylor; Password = 1";
con.Open();
while (true)
{
OracleTransaction t = con.BeginTransaction();
try
{
OracleCommand ora_cmd = new OracleCommand("dequeue_message", con);
ora_cmd.CommandType = CommandType.StoredProcedure;
ora_cmd.Parameters.Add("payload", OracleDbType.Varchar2, ParameterDirection.Output);
ora_cmd.Parameters["payload"].Size = 4000;
ora_cmd.ExecuteNonQuery();
string x = ora_cmd.Parameters["payload"].Value.ToString();
Console.WriteLine(x);
t.Commit();
}
catch (Exception)
{
t.Rollback();
}
}
