Hidden Gems: Event-Driven Change Notifications in Relational Databases

CREATE OR REPLACE FUNCTION notify_change() RETURNS TRIGGER AS $$

  BEGIN

    --

    -- WARNING: Case is VERY IMPORTANT here! If we use 'exampleChannel' PG converts this to

    --          examplechannel and no events will be received!!

    --

    --  UPDATE: [to be confirmed] Case can be handled in PostgreSQL by using double quotes.

    --

    --          In theory, if you had the following line as the listener, it would work in camelCase.

    --

    --          statement.execute('LISTEN "exampleChannel"');

    --

    --          The same applies to any identifier in Postgres.

    --

    PERFORM pg_notify('examplechannel', NEW.phrase);

    RETURN NEW;

  END;

$$ LANGUAGE plpgsql;

CREATE TRIGGER table_change

  AFTER INSERT OR UPDATE OR DELETE ON example

  FOR EACH ROW EXECUTE PROCEDURE notify_change();

@Grapes(

    @Grab(group='com.impossibl.pgjdbc-ng', module='pgjdbc-ng', version='0.8.4')

)

import com.impossibl.postgres.api.jdbc.PGConnection

import com.impossibl.postgres.api.jdbc.PGNotificationListener

import com.impossibl.postgres.jdbc.PGDataSource

PGDataSource dataSource = new PGDataSource();

dataSource.setHost("0.0.0.0")

dataSource.setPort(5432)

dataSource.setDatabaseName("testdb")

dataSource.setUser("postgres")

dataSource.setPassword("password")

final def pgNotificationListener = new PGNotificationListener () { 

  @Override

  public void notification(int processId, String channelName, String payload) {

    println("processId $processId, channelName: $channelName, payload: $payload")

  }

}

final def connection = (PGConnection) dataSource.getConnection()

connection.addNotificationListener(pgNotificationListener)

final def statement = connection.createStatement()

statement.execute("LISTEN examplechannel")

statement.close()

def time = 60 * 60 * 1000

println "Will sleep for $time milliseconds..."

try {

  Thread.sleep (time)

} catch (Throwable thrown) {

  thrown.printStackTrace (System.err)

} finally {

  connection.close ()

}

print "...done!

#

# In this example Docker is running on another machine so assume that I've ssh'd into that box and

# am running the following on the remote machine.

#

docker run -d -p 1521:1521 oracleinanutshell/oracle-xe-11g

docker exec -it [container id] /bin/sh

su

#

# Username: system, password: oracle

#

/u01/app/oracle/product/11.2.0/xe/bin/sqlplus

--

-- This is required otherwise notifications won't be sent to the JDBC driver.

--

grant change notification to system;

commit;

CREATE TABLE example(

  example_id NUMBER(10) PRIMARY KEY,

  phrase VARCHAR2(120) NOT NULL

);

commit;

insert into example values (1, 'one');

insert into example values (2, 'two');

insert into example values (3, 'three');

insert into example values (4, 'four');

insert into example values (5, 'five');

commit;

--

-- Then once the DatabaseChangeListenerInOracleDatabaseExample.groovy is running

-- execute the following and an update should appear in the Groovy console:

--

update example set phrase = 'one / 1' where example_id = 1;

@GrabConfig(systemClassLoader=true)

//

// https://docs.oracle.com/cd/E11882_01/appdev.112/e13995/index.html?oracle/jdbc/dcn/DatabaseChangeRegistration.html

//

// https://mvnrepository.com/artifact/com.oracle.database.jdbc/ojdbc6

@Grapes(

    @Grab(group='com.oracle.database.jdbc', module='ojdbc6', version='11.2.0.4')

)

import oracle.jdbc.dcn.DatabaseChangeListener

import oracle.jdbc.dcn.DatabaseChangeEvent

import oracle.jdbc.driver.OracleConnection

import oracle.jdbc.dcn.DatabaseChangeRegistration

import oracle.jdbc.OracleStatement

import java.sql.DriverManager

import java.util.Properties

//

// Note that the thin driver supports this example.

//

//

// SEE THE WARNING BELOW ABOUT RUNNING THIS SCRIPT ON LOCALHOST WITH ORACLE DB IN DOCKER, ALSO ON LOCALHOST.

//

final def connection = DriverManager.getConnection("jdbc:oracle:thin:@192.168.1.232:1521:xe", "system", "oracle")

def databaseProperties = new Properties ()

/*

 * [6] When the notification type is OCN, any DML transaction that changes one or more registered objects generates

 *     one notification for each object when it commits.

 *

 *     When the notification type is QRCN, any DML transaction that changes the result of one or more registered

 *     queries generates a notification when it commits. The notification includes the query IDs of the queries whose

 *     results changed.

 *

 *     For either notification type, the notification includes:

 *

 *     Name of each changed table

 *

 *     Operation type (INSERT, UPDATE, or DELETE)

 *

 *     ROWID of each changed row, if the registration was created with the ROWID option and the number of modified rows

 *     was not too large. For more information, see ROWID Option."

 */

databaseProperties.setProperty(OracleConnection.DCN_NOTIFY_ROWIDS, "true")

databaseProperties.setProperty(OracleConnection.DCN_QUERY_CHANGE_NOTIFICATION, "true")

final def databaseChangeRegistration = connection.registerDatabaseChangeNotification(databaseProperties)

public class ExampleDatabaseChangeListener implements DatabaseChangeListener {

    @Override

    public void onDatabaseChangeNotification(DatabaseChangeEvent databaseChangeEvent) {

        println ("***** databaseChangeEvent: $databaseChangeEvent")

        println ("***** databaseChangeEvent.source: ${databaseChangeEvent.source}")

        println ("***** databaseChangeEvent.queryChangeDescription: ${databaseChangeEvent.queryChangeDescription}")

        println ("***** databaseChangeEvent.tableChangeDescription: ${databaseChangeEvent.tableChangeDescription.each {println '\n  - nextTableChangeDescription: $it' } }")

    }

}

databaseChangeRegistration.addListener(new ExampleDatabaseChangeListener ())

final def statement = connection.createStatement()

statement.setDatabaseChangeRegistration(databaseChangeRegistration)

try {

  resultSet = statement.executeQuery("select * from example")

  while (resultSet.next())

    {} // println "resultSet.phrase: ${resultSet.getString('phrase')}"

} catch (Throwable thrown) {

  thrown.printStackTrace (System.err)

}

println "databaseChangeRegistration.userName: ${databaseChangeRegistration.userName}"

databaseChangeRegistration.tables.each {

    println "tables: $it"

}

final def time = 60 * 60 * 1000

println "Will sleep for $time milliseconds..."

try {

  Thread.sleep (time)

} catch (Throwable thrown) {

  thrown.printStackTrace (System.err)

} finally {

  statement.close ()

  connection.close ()

}

println "...done!"

/* WARNING: I'm copy-pasting the below message because this is important when running Oracle in Docker and then

 *          running this script on localhost. This caused me a few hours of time trying to figure out why the

 *          notification wasn't being received and ONLY APPLIES IF YOU'RE RUNNING DOCKER ON THE SAME MACHINE AS THIS

 *          SCRIPT IS BEING EXECUTED ON! In fact, I'm not bothering with this at the moment and am running Docker with

 *          Oracle on another machine entirely.

 *

 *          Note also that I've not been able to get this running ON THE SAME MACHINE using:

 *

 *          docker run -d -p 1521:1521 -p [47632:47632] oracleinanutshell/oracle-xe-11g

 *

 * FROM:

 *

 * https://stackoverflow.com/questions/26003506/databasechangeregistration-in-remote-server

 *

 * "You can check active listeners in the Oracle database running the following query:

 *

 * Select * FROM USER_CHANGE_NOTIFICATION_REGS

 * I the query does not return any rows probably the database server can't access the jdbc driver listener port.

 *

 * By default the Oracle JDBC driver listens at port 47632 for notifications. You will need to ensure that it is possible to connect to that port from the database server. You may need to add a rule in the firewall to accept incoming requests to that port.

 *

 * This port can be changed with the NTF_LOCAL_TCP_PORT option:

 *

 * prop.setProperty(OracleConnection.NTF_LOCAL_TCP_PORT, "15000");"

 *

 */

@GrabConfig(systemClassLoader=true)

@Grapes(

  @Grab(group="com.h2database", module="h2", version="1.4.200")

)

import org.h2.api.DatabaseEventListener

import java.sql.SQLException

import java.sql.DriverManager

public class ExampleDatabaseEventListener implements DatabaseEventListener {

  public void closingDatabase () {

    println "closingDatabase: method invoked."

  }

  public void exceptionThrown (SQLException sqle, String sql) {

    println "exceptionThrown: method invoked; sqle: $sqle, sql: $sql"

  }

  public void init (String url) {

    println "init: method invoked; url: $url"

  }

  public void opened () {

    println "opened: method invoked."

  }

  public void setProgress (int state, String name, int x, int max) {

    println "setProgress: method invoked; state: $state, name: $name, x: $x, max: $max"

  }

}

//

// Note the event listener has been added as a parameter in the connection URL below.

//

def cxn = DriverManager.getConnection("jdbc:h2:mem:EventListenerInH2DatabaseExampleDB;DB_CLOSE_DELAY=-1;DATABASE_EVENT_LISTENER='ExampleDatabaseEventListener';")

def stm = cxn.createStatement()

def resultSet = stm.executeQuery("SELECT 1+1")

if (resultSet.next()) {

  println("next: ${resultSet.getInt(1)}")

}

cxn.close ()

println "...Done!"