We have already created our streams administrator stradmin, now we connect with the owner of the source_table to grant him necessary privileges.
conn owner/pass
grant all on owner.source_table to stradmin;
we create a second table (source_table_audit) the same as the source table to insert all changes made in the source table, we create this table with 3 extra columns
- update_date =====> date of the dml
- username =====> the user who made the modification
- action =====> dml (delete, update or insert)
We grant all necessary privileges to the streams administrator on this table
grant all on owner.source_table_audit to stradmin;
We create a queue for this purpose
BEGIN
DBMS_STREAMS_ADM.SET_UP_QUEUE (
queue_table => 'strmadmin.streams_queue_table',
queue_name => 'strmadmin.streams_queue');
END;
/
We create a capture rule to include only dml on our source_table
BEGIN
DBMS_STREAMS_ADM.ADD_TABLE_RULES (
table_name => 'hr.employees',
streams_type => 'capture',
streams_name => 'capture_emp',
queue_name => 'strmadmin.streams_queue',
include_dml => TRUE,
include_ddl => FALSE,
inclusion_rule => TRUE);
END;
/
We add an extra attribute to capture the user who made the change
BEGIN
DBMS_CAPTURE_ADM.INCLUDE_EXTRA_ATTRIBUTE (
capture_name => 'capture_emp',
attribute_name => 'username',
include => TRUE);
END;
/
CREATE OR REPLACE PROCEDURE dml_handler (in_any IN ANYDATA)
IS
lcr SYS.LCR$_ROW_RECORD;
rc PLS_INTEGER;
command VARCHAR2 (30);
old_values SYS.LCR$_ROW_LIST;
BEGIN
-- Access the LCR
rc := in_any.GETOBJECT (lcr);
-- Get the object command
typecommand := lcr.GET_COMMAND_TYPE ();
-- Set the command_type in the row LCR to INSERT
lcr.SET_COMMAND_TYPE ('INSERT');
-- Set the object_name in the row LCR
lcr.SET_OBJECT_NAME ('SOURCE_TABLE_AUDIT');
-- Set the new values to the old values for update and delete
IF command IN ('DELETE', 'UPDATE')
THEN
-- Get the old values in the row LCR
old_values := lcr.GET_VALUES ('old');
-- Set the old values in the row LCR to the new values in the row LCR
lcr.SET_VALUES ('new', old_values);
-- Set the old values in the row LCR to NULL
lcr.SET_VALUES ('old', NULL);
END IF;
-- Add a SYSDATE for update_date
lcr.ADD_COLUMN ('new', 'UPDATE_DATE', ANYDATA.ConvertDate (SYSDATE));
-- Add a user column
lcr.ADD_COLUMN ('new', 'USERNAME', lcr.GET_EXTRA_ATTRIBUTE ('USERNAME'));
-- Add an action column
lcr.ADD_COLUMN ('new', 'ACTION', ANYDATA.ConvertVarChar2 (command));
-- Make the changes
lcr.EXECUTE (TRUE);
COMMIT;
END;
/
Now we can set the apply dml handler for this table
BEGIN
DBMS_APPLY_ADM.SET_DML_HANDLER (
object_name => 'owner.source_table',
object_type => 'TABLE',
operation_name => 'INSERT',
error_handler => FALSE,
user_procedure => 'strmadmin.dml_handler',
apply_database_link => NULL,
apply_name => NULL);
END;
/
BEGIN
DBMS_APPLY_ADM.SET_DML_HANDLER (
object_name => 'owner.source_table',
object_type => 'TABLE',
operation_name => 'UPDATE',
error_handler => FALSE,
user_procedure => 'strmadmin.dml_handler',
apply_database_link => NULL,
apply_name => NULL);
END;
/
BEGIN
DBMS_APPLY_ADM.SET_DML_HANDLER (
object_name => 'owner.source_table',
object_type => 'TABLE',
operation_name => 'DELETE',
error_handler => FALSE,
user_procedure => 'strmadmin.dml_handler',
apply_database_link => NULL,
apply_name => NULL);
END;
/
