/* * CDDL HEADER START * * The contents of this file are subject to the terms of the * Common Development and Distribution License, Version 1.0 only * (the "License"). You may not use this file except in compliance * with the License. * * You can obtain a copy of the license at * docs/licenses/cddl.txt * or http://www.opensource.org/licenses/cddl1.php. * See the License for the specific language governing permissions * and limitations under the License. * * When distributing Covered Code, include this CDDL HEADER in each * file and include the License file at * docs/licenses/cddl.txt. If applicable, * add the following below this CDDL HEADER, with the fields enclosed * by brackets "[]" replaced with your own identifying information: * Portions Copyright [yyyy] [name of copyright owner] * * CDDL HEADER END * * * Copyright 2010-2019 Ping Identity Corporation */ package com.unboundid.directory.sdk.examples; import java.sql.PreparedStatement; import java.sql.ResultSet; import java.sql.SQLException; import java.sql.Timestamp; import java.util.ArrayList; import java.util.Iterator; import java.util.LinkedList; import java.util.List; import java.util.Map; import java.util.concurrent.BlockingQueue; import java.util.concurrent.atomic.AtomicLong; import java.io.Serializable; import com.unboundid.ldap.sdk.Attribute; import com.unboundid.ldap.sdk.Entry; import com.unboundid.ldap.sdk.DN; import com.unboundid.util.args.ArgumentParser; import com.unboundid.util.args.ArgumentException; import com.unboundid.util.StaticUtils; import com.unboundid.directory.sdk.sync.api.JDBCSyncSource; import com.unboundid.directory.sdk.sync.config.JDBCSyncSourceConfig; import com.unboundid.directory.sdk.sync.types.SyncServerContext; import com.unboundid.directory.sdk.sync.types.SetStartpointOptions; import com.unboundid.directory.sdk.sync.types.DatabaseChangeRecord; import com.unboundid.directory.sdk.sync.types.DatabaseChangeRecord.ChangeType; import com.unboundid.directory.sdk.sync.types.TransactionContext; import com.unboundid.directory.sdk.sync.types.SyncOperation; import com.unboundid.directory.sdk.sync.util.ScriptUtils; import com.unboundid.directory.sdk.common.api.MonitorProvider; import com.unboundid.directory.sdk.common.types.LogSeverity; import com.unboundid.directory.sdk.common.types.RegisteredMonitorProvider; /** * This class provides a simple example of a JDBC Sync Source which will detect * changes from a single database table and return them to the Synchronization * Server so that they can be synchronized to a destination endpoint. * This implementation does not use any configuration arguments. * <p> * This example also demonstrates how to register an ad-hoc monitor provider * with the server, without having to add any server-side configuration. */ public final class ExampleJDBCSyncSource extends JDBCSyncSource { //The server context which can be used for obtaining the server state, //logging, etc. private SyncServerContext serverContext; //The name of the data table. private static final String DATA_TABLE = "DataTable"; //The name of the changelog table. private static final String CHANGELOG_TABLE = "ChangeLog"; //Handle to the monitor provider registered with the server. private RegisteredMonitorProvider monitorProvider; //Used to keep track of which changes have been retrieved. private volatile long nextChangeNumberToRetrieve; //Used to keep track of which changes have finished processing. //NOTE: this is the official "startpoint" for this implementation. private volatile long lastCompletedChangeNumber; /** * Retrieves a human-readable name for this extension. * * @return A human-readable name for this extension. */ @Override public String getExtensionName() { return "Example JDBC Sync Source"; } /** * Retrieves a human-readable description for this extension. Each element * of the array that is returned will be considered a separate paragraph in * generated documentation. * * @return A human-readable description for this extension, or {@code null} * or an empty array if no description should be available. */ @Override public String[] getExtensionDescription() { return new String[] { "This extension implements a JDBC Sync Source which can be used " + "to detect and synchronize changes from a single data table. " + "For the sake of simplicity, the columns in the table have " + "LDAP attribute names (cn, sn, givenname, etc). The database " + "uses a changelog table to capture changes, and this extension " + "is configured to manage it." }; } /** * Updates the provided argument parser to define any configuration arguments * which may be used by this extension. The argument parser may also be * updated to define relationships between arguments (e.g. to specify * required, exclusive, or dependent argument sets). * * @param parser The argument parser to be updated with the configuration * arguments which may be used by this extension. * * @throws ArgumentException If a problem is encountered while updating the * provided argument parser. */ @Override public void defineConfigArguments(final ArgumentParser parser) throws ArgumentException { // No arguments will be allowed by default. } /** * This hook is called when a Sync Pipe first starts up, when the * <i>resync</i> process first starts up, or when the set-startpoint * subcommand is called from the <i>realtime-sync</i> command line tool. * Any initialization of this sync source should be performed here. This * method should generally store the {@link SyncServerContext} in a class * member so that it can be used elsewhere in the implementation. * * @param ctx * a TransactionContext which provides a valid JDBC connection to the * database. * @param serverContext A handle to the server context for the server in * which this extension is running. * @param config The general configuration for this sync source. * @param parser The argument parser which has been initialized from * the configuration for this JDBC sync source. */ @Override public void initializeJDBCSyncSource(final TransactionContext ctx, final SyncServerContext serverContext, final JDBCSyncSourceConfig config, final ArgumentParser parser) { this.serverContext = serverContext; this.monitorProvider = serverContext.registerMonitorProvider( new SimpleMonitorProvider(), config); } /** * This hook is called when a Sync Pipe shuts down, when the <i>resync</i> * process shuts down, or when the set-startpoint subcommand (from the * <i>realtime-sync</i> command line tool) is finished. Any clean up of this * sync source should be performed here. * * @param ctx * a TransactionContext which provides a valid JDBC connection to the * database. */ @Override public void finalizeJDBCSyncSource(final TransactionContext ctx) { serverContext.deregisterMonitorProvider(monitorProvider); } /** * This method should effectively set the starting point for synchronization * to the place specified by the <code>options</code> parameter. This should * cause all changes previous to the specified start point to be disregarded * and only changes after that point to be returned by * {@link #getNextBatchOfChanges(TransactionContext, int, AtomicLong)}. * <p> * There are several different startpoint types (see * {@link SetStartpointOptions}), and this implementation is not required to * support them all. If the specified startpoint type is unsupported, this * method should throw an {@link IllegalArgumentException}. * <p> * <b>IMPORTANT</b>: The <code>RESUME_AT_SERIALIZABLE</code> startpoint type * must be supported by your implementation, because this is used when a Sync * Pipe first starts up. * <p> * This method can be called from two different contexts: * <ul> * <li>When the 'set-startpoint' subcommand of the realtime-sync CLI is used * (the Sync Pipe is required to be stopped in this context)</li> * <li>Immediately after a connection is first established to the source * server (e.g. before the first call to * {@link #getNextBatchOfChanges(TransactionContext, int, AtomicLong)})</li> * </ul> * @param ctx * a TransactionContext which provides a valid JDBC connection to the * database. * @param options * an object which indicates where exactly to start synchronizing * (e.g. the end of the changelog, specific change number, a certain * time ago, etc) * @throws SQLException * if there is any error while setting the start point */ @Override public void setStartpoint(final TransactionContext ctx, final SetStartpointOptions options) throws SQLException { switch(options.getStartpointType()) { case BEGINNING_OF_CHANGELOG: lastCompletedChangeNumber = 0; nextChangeNumberToRetrieve = 0; break; case END_OF_CHANGELOG: PreparedStatement stmt = ctx.prepareStatement( "SELECT MAX(change_number) AS value FROM " + CHANGELOG_TABLE); ResultSet rset = stmt.executeQuery(); try { long value = 0; if(rset.next()) { value = rset.getLong("value"); } else { String msg = "Could not find max change number"; serverContext.logMessage(LogSeverity.SEVERE_ERROR, msg); throw new SQLException(msg); } lastCompletedChangeNumber = value; nextChangeNumberToRetrieve = value + 1; } finally { rset.close(); stmt.close(); } break; case RESUME_AT_CHANGE_NUMBER: nextChangeNumberToRetrieve = options.getChangeNumber(); lastCompletedChangeNumber = nextChangeNumberToRetrieve - 1; break; case RESUME_AT_SERIALIZABLE: //When sync first starts up, this method is //called with this StartpointType to //initialize the internal state. Serializable token = options.getSerializableValue(); if(token != null) { lastCompletedChangeNumber = (Long) token; nextChangeNumberToRetrieve = lastCompletedChangeNumber + 1; } break; default: throw new IllegalArgumentException( "This startpoint type is not supported: " + options.getStartpointType().toString()); } } /** * Gets the current value of the startpoint for change detection. This is the * "bookmark" which indicates which changes have already been processed and * which have not. In most cases, a change number is used to detect changes * and is managed by the Data Sync Server, in which case this * implementation needs only to return the latest acknowledged * change number. In other cases, the return value may correspond to a * different value, such as the SYS_CHANGE_VERSION in Microsoft SQL Server. * In any case, this method should return the value that is updated by * {@link #acknowledgeCompletedOps(TransactionContext, LinkedList)}. * <p> * This method is called periodically and the return value is saved in the * persistent state for the Sync Pipe that uses this script as its Sync * Source. * <p> * <b>IMPORTANT</b>: The internal value for the startpoint should only be * updated after a sync operation is acknowledged back to this script (via * {@link #acknowledgeCompletedOps(TransactionContext, LinkedList)}). * Otherwise it will be possible for changes to be missed when the * Data Sync Server is restarted or a connection error occurs. * @return a value to store in the persistent state for the Sync Pipe. This is * usually a change number, but if a changelog table is not used to * detect changes, this value should represent some other token to * pass into * {@link #setStartpoint(TransactionContext, SetStartpointOptions)} * when the sync pipe starts up. */ @Override public Serializable getStartpoint() { return Long.valueOf(lastCompletedChangeNumber); } /** * Return a full source entry (in LDAP form) from the database, corresponding * to the {@link DatabaseChangeRecord} that is passed in through the * {@link SyncOperation}. This method should perform any queries necessary to * gather the latest values for all the attributes to be synchronized. * <p> * This method <b>must be thread safe</b>, as it will be called repeatedly and * concurrently by each of the Sync Pipe worker threads as they process * entries. * @param ctx * a TransactionContext which provides a valid JDBC connection to the * database. * @param operation * the SyncOperation which identifies the database "entry" to * fetch. The DatabaseChangeRecord can be obtained by calling * <code>operation.getDatabaseChangeRecord()</code>. * This is what is returned by * {@link #getNextBatchOfChanges(TransactionContext, int, AtomicLong)} * and also what comes out of * {@link #listAllEntries(TransactionContext, String, BlockingQueue)}. * @return a full LDAP Entry, or null if no such entry exists. * @throws SQLException * if there is an error fetching the entry */ @Override public Entry fetchEntry(final TransactionContext ctx, final SyncOperation operation) throws SQLException { //Create a map of all the identity key/value pairs (delimited by '%%') that //make up the identifiable info. In this single-table example, the //identifier is always "uid={uid}". This comes from the 'identifier' column //in the changelog table. It is up to the trigger code for each table to //construct this. DatabaseChangeRecord changeRecord = operation.getDatabaseChangeRecord(); DN id = changeRecord.getIdentifiableInfo(); Map<String,String> keyAndValue = ScriptUtils.dnToMap(id); if (! keyAndValue.containsKey("uid")) { throw new SQLException("No value found for uid in " + changeRecord); } Entry entry; String entryType = changeRecord.getEntryType(); if(entryType.equalsIgnoreCase("person")) { long uid = Long.valueOf(keyAndValue.get("uid")); String sql = "SELECT * FROM " + DATA_TABLE + " WHERE uid = ?"; PreparedStatement stmt = ctx.prepareStatement(sql); try { stmt.setLong(1, uid); //Create an entry using the literal column names as attribute names entry = ctx.searchToRawEntry(stmt, "uid"); } finally { stmt.close(); } } else { throw new IllegalArgumentException("Unknown entry type: " + entryType); } return entry; } /** * Provides a way for the Data Sync Server to acknowledge back to the * script which sync operations it has processed. This method should update * the official startpoint which was set by * {@link #setStartpoint(TransactionContext, SetStartpointOptions)} and is * returned by {@link #getStartpoint()}. * <p> * <b>IMPORTANT</b>: The internal value for the startpoint should only be * updated after a sync operation is acknowledged back to this script (via * this method). Otherwise it will be possible for changes to be missed when * the Data Sync Server is restarted or a connection error occurs. * <p> * A {@link TransactionContext} is provided in case the acknowledgment needs * to make it all the way back to the database itself (for example if you were * using Oracle's Change Data Capture). * @param ctx * a TransactionContext which provides a valid JDBC connection to the * database. * @param completedOps * a list of {@link SyncOperation}s that have finished processing. * The records are listed in the order they were first detected. * @throws SQLException * if there is an error acknowledging the changes back to the * database */ @Override public void acknowledgeCompletedOps( final TransactionContext ctx, final LinkedList<SyncOperation> completedOps) throws SQLException { if(!completedOps.isEmpty()) { //Update lastCompletedChangeNumber to that of the last completed operation DatabaseChangeRecord last = completedOps.getLast().getDatabaseChangeRecord(); lastCompletedChangeNumber = last.getChangeNumber(); } } /** * Return the next batch of change records from the database. Change records * are just hints that a change happened; they do not include the actual data * of the change. In an effort to never synchronize stale data, the * Data Sync Server will go back and fetch the full source entry for * each change record. * <p> * On the first invocation, this should return changes starting from the * startpoint that was set by * {@link #setStartpoint(TransactionContext, SetStartpointOptions)}. This * method is responsible for updating the internal state such that subsequent * invocations do not return duplicate changes. * <p> * The resulting list should be limited by <code>maxChanges</code>. The * <code>numStillPending</code> reference should be set to the estimated * number of changes that haven't yet been retrieved from the changelog table * when this method returns, or zero if all the current changes have been * retrieved. * <p> * <b>IMPORTANT</b>: While this method needs to keep track of which changes * have already been returned so that it does not return them again, it should * <b>NOT</b> modify the official startpoint. The internal value for the * startpoint should only be updated after a sync operation is acknowledged * back to this script (via * {@link #acknowledgeCompletedOps(TransactionContext, LinkedList)}). * Otherwise it will be possible for changes to be missed when the * Data Sync Server is restarted or a connection error occurs. The * startpoint should not change as a result of this method. * <p> * This method <b>does not need to be thread-safe</b>. It will be invoked * repeatedly by a single thread, based on the polling interval set in the * Sync Pipe configuration. * @param ctx * a TransactionContext which provides a valid JDBC connection to the * database. * @param maxChanges * the maximum number of changes to retrieve * @param numStillPending * this should be set to the number of unretrieved changes that * are still pending after this batch has been retrieved. This will * be passed in as zero, and may be left that way if the actual value * cannot be determined. * @return a list of {@link DatabaseChangeRecord} instances, each * corresponding to a row in the changelog table (or the equivalent if * some other change tracking mechanism is being used). If there are * no new changes to return, this method should return an empty list. * @throws SQLException * if there is any error while retrieving the next batch of changes */ @Override public List<DatabaseChangeRecord> getNextBatchOfChanges( final TransactionContext ctx, final int maxChanges, final AtomicLong numStillPending) throws SQLException { List<DatabaseChangeRecord> results = new ArrayList<DatabaseChangeRecord>(); PreparedStatement stmt = ctx.prepareStatement( "SELECT * FROM " + CHANGELOG_TABLE + " WHERE change_number >= ? ORDER BY change_number ASC"); //This is a generic way to limit the size of the result set; however, this //may or may not be implemented by the JDBC driver to alter the query //itself. It may just allow the full set of results to come back and then //perform the truncation within the JVM, which can cause out of memory //errors. Most databases have a much more efficient mechansim (for example //"SELECT TOP(1000)..." in SQLServer and "SELECT ... WHERE ROWNUM < 1000" //in Oracle). stmt.setMaxRows(maxChanges); stmt.setLong(1, nextChangeNumberToRetrieve); ResultSet rset = stmt.executeQuery(); while(rset.next()) { if(results.size() >= maxChanges) { serverContext.debugError( "The result set contained too many rows; expected no more than " + maxChanges); break; } //In this case there is a change_type column in the changelog table which //gives us the change type ChangeType type = ChangeType.valueOf(rset.getString("change_type")); DatabaseChangeRecord.Builder bldr = new DatabaseChangeRecord.Builder( type, rset.getString("identifier")); long changeNum = rset.getLong("change_number"); //Update nextChangeNumberToRetrieve so that the next call will get the //next batch nextChangeNumberToRetrieve = changeNum + 1; bldr.changeNumber(changeNum); bldr.tableName(rset.getString("table_name")); String entryType = rset.getString("entry_type"); if(!entryType.equalsIgnoreCase("person")) { //This sync source only handles the "person" entry type serverContext.debugInfo("Skipping change with entry type: " + entryType); continue; } bldr.entryType(entryType); //Get the list of changed columns for this change. For UPDATE operations, //the Data Sync Server will only modify the destination //attributes that depend on the originally changed source columns (i.e. //if this is not set, no modifications will take place to the destination //entry). String cols = rset.getString("changed_columns"); bldr.changedColumns(cols != null ? cols.split(",") : null); //Get the database user who made the change bldr.modifier(rset.getString("modifiers_name")); //Get the timestamp of the change bldr.changeTime(rset.getTimestamp("change_time").getTime()); results.add(bldr.build()); } rset.close(); stmt.close(); //Figure out how many changes are still unretrieved at this point stmt = ctx.prepareStatement("SELECT COUNT(*) FROM " + CHANGELOG_TABLE + " WHERE change_number >= ?"); stmt.setLong(1, nextChangeNumberToRetrieve); rset = stmt.executeQuery(); if(rset.next()) { long stillPending = rset.getLong(1); numStillPending.set(stillPending); } rset.close(); stmt.close(); return results; } /** * Gets a list of all the entries in the database for a given entry type. This * is used by the 'resync' command line tool. The default implementation * throws a {@link UnsupportedOperationException}; subclasses should override * if the resync functionality is needed. * <p> * The <code>entryType</code> is user-defined; it will be * passed in on the command line for resync. The <code>outputQueue</code> * should contain {@link DatabaseChangeRecord} objects with the * <code>ChangeType</code> set to <i>resync</i>. * <p> * This method should not return until all the entries of the given entryType * have been added to the output queue. Separate threads will concurrently * drain entries from the queue and process them. (The queue should not * actually contain full entries, but rather DatabaseChangeRecord objects * which identify the full database entries. These objects are then * individually passed in to * {@link #fetchEntry(TransactionContext, SyncOperation)}. Therefore, * it is important to make sure that the DatabaseChangeRecord instances * contain enough identifiable information (e.g. primary keys) for each entry * so that the entry can be found again. * <p> * The lifecycle of resync is similar to that of real-time sync, with a few * differences: * <ol> * <li>Stream out a list of all IDs in the database (for a given entryType) * </li> * <li>Fetch full source entry for an ID</li> * <li>Perform any mappings and compute the equivalent destination entry</li> * <li>Fetch full destination entry</li> * <li>Diff the computed destination entry and actual destination entry</li> * <li>Apply the minimal set of changes at the destination to bring it in sync * </li> * </ol> * If the total set of entries is very large, it is fine to split up the work * into multiple database queries within this method. The queue will not grow * out of control because it blocks when it becomes full. The queue capacity * is fixed at 1000. * <p> * @param ctx * a TransactionContext which provides a valid JDBC connection to the * database. * @param entryType * the type of database entry to be fetched (this is specified * on the CLI for the resync command) * @param outputQueue * a queue of DatabaseChangeRecord objects which will be individually * fetched via * {@link #fetchEntry(TransactionContext, SyncOperation)} * @throws SQLException * if there is an error retrieving the list of entries to resync */ @Override public void listAllEntries( final TransactionContext ctx, final String entryType, final BlockingQueue<DatabaseChangeRecord> outputQueue) throws SQLException { serverContext.debugInfo("Beginning to dump all entries..."); if(entryType.equalsIgnoreCase("person")) { //Get a full list of the UIDs PreparedStatement stmt = ctx.prepareStatement( "SELECT uid FROM " + DATA_TABLE + " ORDER BY uid ASC"); ResultSet rset = stmt.executeQuery(); while(rset.next()) { long uid = rset.getLong("uid"); DatabaseChangeRecord.Builder bldr = new DatabaseChangeRecord.Builder(ChangeType.resync, "uid=" + uid); //set the entry type so that fetchEntry() can use it bldr.entryType(entryType); try { outputQueue.put(bldr.build()); } catch(InterruptedException e) { serverContext.logMessage(LogSeverity.SEVERE_ERROR, StaticUtils.getExceptionMessage(e)); rset.close(); stmt.close(); throw new SQLException("The outputQueue was interrupted.", e); } } rset.close(); stmt.close(); } else { throw new IllegalArgumentException("Unknown entry type: " + entryType); } } /** * Gets a list of all the entries in the database from a given file input. * This is used by the 'resync' command line tool. The default implementation * throws a {@link UnsupportedOperationException}; subclasses should override * if the resync functionality is needed for specific database records, which * can be specified in the input file. * <p> * The format for the <code>inputLines</code> (e.g. the content of the file) * is user-defined; it may be key/value pairs, primary keys, or full SQL * statements, for example. The use of this method is triggered via the * <i>--sourceInputFile</i> argument on the resync CLI. The * <code>outputQueue</code> should contain {@link DatabaseChangeRecord} * objects with the <code>ChangeType</code> set to <i>resync</i>. * <p> * This method should not return until all the entries specified by the input * file have been added to the output queue. Separate threads will * concurrently drain entries from the queue and process them. (The queue * should not actually contain full entries, but rather DatabaseChangeRecord * objects which identify the full database entries. These objects are then * individually passed in to * {@link #fetchEntry(TransactionContext, SyncOperation)}. Therefore, * it is important to make sure that the DatabaseChangeRecord instances * contain enough identifiable information (e.g. primary keys) for each entry * so that the entry can be found again. * <p> * The lifecycle of resync is similar to that of real-time sync, with a few * differences: * <ol> * <li>Stream out a list of all IDs in the database (using the given input * file)</li> * <li>Fetch full source entry for an ID</li> * <li>Perform any mappings and compute the equivalent destination entry</li> * <li>Fetch full destination entry</li> * <li>Diff the computed destination entry and actual destination entry</li> * <li>Apply the minimal set of changes at the destination to bring it in sync * </li> * </ol> * If the total set of entries is very large, it is fine to split up the work * into multiple database queries within this method. The queue will not grow * out of control because it blocks when it becomes full. The queue capacity * is fixed at 1000. * <p> * @param ctx * a TransactionContext which provides a valid JDBC connection to the * database. * @param entryType * the type of database entry to be fetched (this is specified * on the CLI for the resync command) * @param inputLines * an Iterator containing the lines from the specified input file to * resync (this is specified on the CLI for the resync command). * These lines can be any format, for example a set of primary keys, * a set of WHERE clauses, a set of full SQL queries, etc. * @param outputQueue * a queue of DatabaseChangeRecord objects which will be individually * fetched via {@link #fetchEntry(TransactionContext, SyncOperation)} * @throws SQLException * if there is an error retrieving the list of entries to resync */ @Override public void listAllEntries( final TransactionContext ctx, final String entryType, final Iterator<String> inputLines, final BlockingQueue<DatabaseChangeRecord> outputQueue) throws SQLException { if(!entryType.equalsIgnoreCase("person")) { throw new IllegalArgumentException("Unknown entry type: " + entryType); } while(inputLines.hasNext()) { String uid = inputLines.next().trim(); if(uid.isEmpty()) { continue; } DatabaseChangeRecord.Builder bldr = new DatabaseChangeRecord.Builder( ChangeType.resync, "uid=" + uid); //Set the entry type so that fetchEntry() can use it bldr.entryType(entryType); try { outputQueue.put(bldr.build()); } catch(InterruptedException e) { serverContext.logMessage(LogSeverity.SEVERE_ERROR, StaticUtils.getExceptionMessage(e)); throw new SQLException("The outputQueue was interrupted.", e); } } } /** * Performs a cleanup of the changelog table (if desired). There is a * background thread that periodically invokes this method. It should remove * any rows in the changelog table that are more than * <code>maxAgeMillis</code> milliseconds old. * <p> * <b>NOTE:</b> If the system clock on the database server is not in sync with * the system clock on the Data Sync Server, this method should query * the database for its current time in order to determine the cut-off point * for deleting changelog records. * <p> * If a separate mechanism will be used to manage the changelog table, this * method may be implemented as a no-op and always return zero. This is how * the default implementation behaves. * @param ctx * a TransactionContext which provides a valid JDBC connection to the * database. * @param maxAgeMillis * the period of time (in milliseconds) after which a changelog table * record should be deleted * @return the number of rows that were deleted from the changelog table * @throws SQLException * if there is an error purging records from the changelog table */ @Override public int cleanupChangelog(final TransactionContext ctx, final long maxAgeMillis) throws SQLException { //get current time on database PreparedStatement stmt = ctx.prepareStatement("SELECT CURRENT_TIMESTAMP"); ResultSet rset = stmt.executeQuery(); long currentTimeMillis; try { if(rset.next()) { currentTimeMillis = rset.getTimestamp(1).getTime(); } else { throw new SQLException( "Cannot determine current timestamp on database."); } } finally { rset.close(); stmt.close(); } stmt = ctx.prepareStatement( "DELETE FROM " + CHANGELOG_TABLE + " WHERE change_time < ?"); stmt.clearWarnings(); stmt.setTimestamp(1, new Timestamp(currentTimeMillis - maxAgeMillis)); int rowCount = stmt.executeUpdate(); stmt.close(); return rowCount; } /** * A simple implementation of MonitorProvider that we can register with the * server through the ServerContext object. This provides the current values * of nextChangeNumberToRetrieve and lastCompletedChangeNumber. */ private class SimpleMonitorProvider extends MonitorProvider { /** * Retrieves a human-readable name for this extension. This is not used * since we are registering this MonitorProvider through the ServerContext * object (e.g. it is not a stand-alone extension). * * @return A human-readable name for this extension. */ @Override public String getExtensionName() { return "SimpleMonitorProvider"; } /** * Retrieves a human-readable description for this extension. Each element * of the array that is returned will be considered a separate paragraph in * generated documentation. This is not used since we are registering this * MonitorProvider through the ServerContext object (e.g. it is not a * stand-alone extension). * * @return A human-readable description for this extension, or {@code null} * or an empty array if no description should be available. */ @Override public String[] getExtensionDescription() { return null; } /** * Retrieves the name that identifies this monitor provider instance. It * will be used as the value of the naming attribute for monitor entries. * Each monitor provider instance must have a unique name. * * @return The name that identifies this monitor provider instance. */ @Override public String getMonitorInstanceName() { return "ExampleJDBCSyncSource Monitor"; } /** * Retrieves the name of the object class that will be used for monitor * entries created from this monitor provider. This does not need to be * defined in the server schema. It may be {@code null} if a default object * class should be used. * * @return The name of the object class that will be used for monitor * entries created from this monitor provider. */ @Override public String getMonitorObjectClass() { return "jdbc-sync-source-monitor-entry"; } /** * Retrieves a list of attributes containing the data to include in the * monitor entry generated from this monitor provider. * * @return A list of attributes containing the data to include in the * monitor entry generated from this monitor provider. */ @Override public List<Attribute> getMonitorAttributes() { List<Attribute> attributes = new ArrayList<Attribute>(2); attributes.add(new Attribute("next-changenumber-to-retrieve", String.valueOf(nextChangeNumberToRetrieve))); attributes.add(new Attribute("last-completed-change-number", String.valueOf(lastCompletedChangeNumber))); return attributes; } } }