* 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(replicationServer, dbenv); firstkey = db.readFirstDraftCN(); lastkey = db.readLastDraftCN(); // Trimming thread thread = new DirectoryThread(this, "Replication DraftCN db"); thread.start(); // Monitoring registration DirectoryServer.deregisterMonitorProvider(dbMonitor); DirectoryServer.registerMonitorProvider(dbMonitor); } /** * Add an update to the list of messages that must be saved to the db * managed by this db handler. * This method is blocking if the size of the list of message is larger * than its maximum. * @param key The key for this record in the db. * @param value The associated value. * @param baseDN The associated baseDN. * @param cn The associated replication change number. */ public synchronized void add(int key, String value, String baseDN, ChangeNumber cn) { db.addEntry(key, value, baseDN, cn); if (debugEnabled()) TRACER.debugInfo( "In DraftCNDbhandler.add, added: " + " key=" + key + " value=" + value + " baseDN=" + baseDN + " cn=" + cn); } /** * Get the firstChange. * @return Returns the firstChange. */ public int getFirstKey() { return db.readFirstDraftCN(); } /** * Get the lastChange. * @return Returns the lastChange. */ public int getLastKey() { return db.readLastDraftCN(); } /** * Get the number of changes. * @return Returns the number of changes. */ public long count() { return db.count(); } /** * Returns whether this database is empty. *
* FIXME Find a way to implement this method in a more efficient manner. * {@link com.sleepycat.je.Database#count()} javadoc mentions: *
Note that this method does scan a significant portion of the * database and should be considered a fairly expensive * operation.*
* It could be faster to: *
db.readFirstDraftCN() != 0true if this database is empty, false
* otherwise
*/
public boolean isEmpty()
{
return count() == 0;
}
/**
* Get a read cursor on the database from a provided key.
* The cursor MUST be released after use.
* @param key The provided key.
* @return the new cursor.
*/
public DraftCNDBCursor getReadCursor(int key)
{
try
{
return db.openReadCursor(key);
}
catch(Exception e)
{
return null;
}
}
/**
* Release a provided read cursor.
* @param cursor The provided read cursor.
*/
public void releaseReadCursor(DraftCNDBCursor cursor)
{
close(cursor);
}
/**
* Generate a new ReplicationIterator that allows to browse the db
* managed by this dbHandler and starting at the position defined
* by a given changeNumber.
*
* @param startDraftCN The position where the iterator must start.
*
* @return a new ReplicationIterator that allows to browse the db
* managed by this dbHandler and starting at the position defined
* by a given changeNumber.
*
* @throws ChangelogException if a database problem happened.
*/
public DraftCNDbIterator generateIterator(int startDraftCN)
throws ChangelogException
{
return new DraftCNDbIterator(db, startDraftCN);
}
/**
* Shutdown this dbHandler.
*/
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(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);
}
/**
* Clear the changes from this DB (from both memory cache and DB storage)
* for the provided baseDN.
* @param baseDNToClear The baseDN for which we want to remove
* all records from the DraftCNDb - null means all.
* @throws ChangelogException When an exception occurs while removing the
* changes from the DB.
*/
public void clear(String baseDNToClear) throws ChangelogException
{
if (isEmpty())
{
return;
}
ChangeNumber crossDomainEligibleCN = replicationServer.getEligibleCN();
for (int i = 0; i < 100; i++)
{
DraftCNDBCursor cursor = db.openDeleteCursor();
try
{
for (int j = 0; j < 50; j++)
{
// let's traverse the DraftCNDb
if (!cursor.next())
{
cursor.close();
return;
}
ChangeNumber cn = cursor.currentChangeNumber();
// From the draftCNDb change record, get the domain and changeNumber
String baseDN = cursor.currentBaseDN();
if ((baseDNToClear != null)
&& (baseDNToClear.equalsIgnoreCase(baseDN)))
{
cursor.delete();
continue;
}
ReplicationServerDomain domain = replicationServer
.getReplicationServerDomain(baseDN, false);
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;
}
ServerState startState = domain.getStartState();
// We don't use the returned endState but it's updating CN as
// reading
domain.getEligibleState(crossDomainEligibleCN);
ChangeNumber fcn = startState.getChangeNumber(cn.getServerId());
int currentKey = cursor.currentKey();
if (cn.older(fcn))
{
cursor.delete();
continue;
}
ServerState cnVector;
try
{
Map