* This class publishes some monitoring information below
* cn=monitor.
*/
public class DraftCNDbHandler implements ChangeNumberIndexDB, Runnable
{
/**
* The tracer object for the debug logger.
*/
private static final DebugTracer TRACER = getTracer();
private static int NO_KEY = 0;
private DraftCNDB db;
/**
* FIXME Is this field that useful? {@link #getFirstChangeNumber()} does not
* even use it!
*/
private long firstChangeNumber = NO_KEY;
/**
* FIXME Is this field that useful? {@link #getLastChangeNumber()} does not
* even use it!
*/
private long lastChangeNumber = NO_KEY;
private DbMonitorProvider dbMonitor = new DbMonitorProvider();
private boolean shutdown = false;
private boolean trimDone = false;
/**
* A dedicated thread loops trim().
*
* trim() : deletes from the DB a number of changes that are older than a * certain date. */ private DirectoryThread thread; /** * The trim age in milliseconds. Changes record in the change DB that are * older than this age are removed. *
* FIXME it never gets updated even when the replication server purge delay is * updated */ private long trimAge; private ReplicationServer replicationServer; /** * Creates a new dbHandler associated to a given LDAP server. * * @param replicationServer The ReplicationServer that creates this dbHandler. * @param dbenv the Database Env to use to create the ReplicationServer DB. * server for this domain. * @throws ChangelogException If a database problem happened */ public DraftCNDbHandler(ReplicationServer replicationServer, ReplicationDbEnv dbenv) throws ChangelogException { this.replicationServer = replicationServer; this.trimAge = replicationServer.getTrimAge(); // DB initialization db = new DraftCNDB(dbenv); firstChangeNumber = getChangeNumber(db.readFirstRecord()); lastChangeNumber = getChangeNumber(db.readLastRecord()); // Trimming thread thread = new DirectoryThread(this, "Replication DraftCN db"); thread.start(); // Monitoring registration DirectoryServer.deregisterMonitorProvider(dbMonitor); DirectoryServer.registerMonitorProvider(dbMonitor); } private long getChangeNumber(CNIndexRecord record) throws ChangelogException { if (record != null) { return record.getChangeNumber(); } return 0; } /** {@inheritDoc} */ @Override public void addRecord(CNIndexRecord record) throws ChangelogException { db.addRecord(record); if (debugEnabled()) TRACER.debugInfo("In DraftCNDbhandler.add, added: " + record); } /** {@inheritDoc} */ @Override public CNIndexRecord getFirstRecord() throws ChangelogException { return db.readFirstRecord(); } /** {@inheritDoc} */ @Override public CNIndexRecord getLastRecord() throws ChangelogException { return db.readLastRecord(); } /** * Get the number of changes. * @return Returns the number of changes. */ public long count() { return db.count(); } /** {@inheritDoc} */ @Override public boolean isEmpty() throws ChangelogException { return getLastRecord() == null; } /** * Get a read cursor on the database from a provided key. The cursor MUST be * closed after use. *
* This method is only used by unit tests.
*
* @param startChangeNumber
* The change number from where to start.
* @return the new cursor.
* @throws ChangelogException
* if a database problem occurs.
*/
DraftCNDBCursor getReadCursor(long startChangeNumber)
throws ChangelogException
{
return db.openReadCursor(startChangeNumber);
}
/** {@inheritDoc} */
@Override
public ChangeNumberIndexDBCursor getCursorFrom(long startChangeNumber)
throws ChangelogException
{
return new DraftCNDbIterator(db, startChangeNumber);
}
/** {@inheritDoc} */
@Override
public void shutdown()
{
if (shutdown)
{
return;
}
shutdown = true;
synchronized (this)
{
notifyAll();
}
synchronized (this)
{ /* Can we just do a thread.join() ? */
while (!trimDone)
{
try
{
wait();
} catch (InterruptedException e)
{ /* do nothing */ }
}
}
db.shutdown();
DirectoryServer.deregisterMonitorProvider(dbMonitor);
}
/**
* Run method for this class.
* Periodically Flushes the ReplicationServerDomain cache from memory to the
* stable storage and trims the old updates.
*/
@Override
public void run()
{
while (!shutdown)
{
try {
trim();
synchronized (this)
{
try
{
wait(1000);
} catch (InterruptedException e)
{
Thread.currentThread().interrupt();
}
}
} catch (Exception end)
{
MessageBuilder mb = new MessageBuilder();
mb.append(ERR_EXCEPTION_CHANGELOG_TRIM_FLUSH.get());
mb.append(" ");
mb.append(stackTraceToSingleLineString(end));
logError(mb.toMessage());
if (replicationServer != null)
replicationServer.shutdown();
break;
}
}
synchronized (this)
{
trimDone = true;
notifyAll();
}
}
/**
* Trim old changes from this database.
* @throws ChangelogException In case of database problem.
*/
public void trim() throws ChangelogException
{
if (trimAge == 0)
return;
clear(null);
}
/** {@inheritDoc} */
@Override
public void clear(DN baseDNToClear) throws ChangelogException
{
if (isEmpty())
{
return;
}
for (int i = 0; i < 100; i++)
{
final DraftCNDBCursor cursor = db.openDeleteCursor();
try
{
for (int j = 0; j < 50; j++)
{
// let's traverse the DraftCNDb
if (!cursor.next())
{
cursor.close();
return;
}
// From the draftCNDb change record, get the domain and CSN
final CNIndexRecord record = cursor.currentRecord();
if (baseDNToClear != null && baseDNToClear.equals(record.getBaseDN()))
{
cursor.delete();
continue;
}
final ReplicationServerDomain domain =
replicationServer.getReplicationServerDomain(record.getBaseDN());
if (domain == null)
{
// the domain has been removed since the record was written in the
// draftCNDb, thus it makes no sense to keep the record in the
// draftCNDb.
cursor.delete();
continue;
}
final CSN csn = record.getCSN();
final ServerState startState = domain.getStartState();
final CSN fcsn = startState.getCSN(csn.getServerId());
final long currentChangeNumber = record.getChangeNumber();
if (csn.older(fcsn))
{
cursor.delete();
continue;
}
ServerState csnVector;
try
{
Map