OPENDJ-1116 Introduce abstraction for the changelog DB

Jean-Noel Rouvignac

30.29.2013 4772c4aec3a23175ad9ac33b4cd843062168d335

OPENDJ-1116 Introduce abstraction for the changelog DB

Renamed:
- ReplicationDBCursor to ReplicaDBCursor
- ReplicationDBCursorComparator to ReplicaDBCursorComparator
- JEReplicationDBCursor to JEReplicaDBCursor

Fixed javadocs, variable names, method names, etc.

 opends/src/server/org/opends/server/replication/server/MessageHandler.java

@@ -281,7 +281,7 @@
           *           restart as usual
           *   load this change on the delayList
           */
          NavigableSet<ReplicationDBCursor> sortedCursors = null;
          NavigableSet<ReplicaDBCursor> sortedCursors = null;
          try
          {
            sortedCursors = collectAllCursorsWithChanges();
@@ -395,7 +395,7 @@
  }

  private UpdateMsg nextOldestUpdateMsg(
      NavigableSet<ReplicationDBCursor> sortedCursors)
      NavigableSet<ReplicaDBCursor> sortedCursors)
  {
    /*
     * The cursors are sorted based on the currentChange of each cursor to
@@ -404,15 +404,15 @@
     * it is necessary to remove and eventually add again a cursor (after moving
     * it forward).
     */
    final ReplicationDBCursor cursor = sortedCursors.pollFirst();
    final ReplicaDBCursor cursor = sortedCursors.pollFirst();
    final UpdateMsg result = cursor.getChange();
    cursor.next();
    addCursorIfNotEmpty(sortedCursors, cursor);
    return result;
  }

  private void addCursorIfNotEmpty(Collection<ReplicationDBCursor> cursors,
      ReplicationDBCursor cursor)
  private void addCursorIfNotEmpty(Collection<ReplicaDBCursor> cursors,
      ReplicaDBCursor cursor)
  {
    if (cursor != null)
    {
@@ -461,16 +461,16 @@
          the lateQueue when it will send the next update but we are not yet
          there. So let's take the last change not sent directly from the db.
          */
          result = findOldestChangeNumberFromReplicationDBs();
          result = findOldestChangeNumberFromReplicaDBs();
        }
      }
    }
    return result;
  }

  private ChangeNumber findOldestChangeNumberFromReplicationDBs()
  private ChangeNumber findOldestChangeNumberFromReplicaDBs()
  {
    SortedSet<ReplicationDBCursor> sortedCursors = null;
    SortedSet<ReplicaDBCursor> sortedCursors = null;
    try
    {
      sortedCursors = collectAllCursorsWithChanges();
@@ -488,16 +488,16 @@
  }

  /**
   * Collects all the replication DB cursors that have changes and sort them
   * Collects all the {@link ReplicaDBCursor}s that have changes and sort them
   * with the oldest {@link ChangeNumber} first.
   *
   * @return a List of cursors with changes sorted by their {@link ChangeNumber}
   *         (oldest first)
   */
  private NavigableSet<ReplicationDBCursor> collectAllCursorsWithChanges()
  private NavigableSet<ReplicaDBCursor> collectAllCursorsWithChanges()
  {
    final NavigableSet<ReplicationDBCursor> results =
        new TreeSet<ReplicationDBCursor>(new ReplicationDBCursorComparator());
    final NavigableSet<ReplicaDBCursor> results =
        new TreeSet<ReplicaDBCursor>(new ReplicaDBCursorComparator());
    for (int serverId : replicationServerDomain.getServerIds())
    {
      // get the last already sent CN from that server to get a cursor

 opends/src/server/org/opends/server/replication/server/ReplicationBackend.java

@@ -50,7 +50,7 @@
import org.opends.server.replication.plugin.MultimasterReplication;
import org.opends.server.replication.plugin.ReplicationServerListener;
import org.opends.server.replication.protocol.*;
import org.opends.server.replication.server.changelog.api.ReplicationDBCursor;
import org.opends.server.replication.server.changelog.api.ReplicaDBCursor;
import org.opends.server.types.*;
import org.opends.server.util.*;

@@ -631,7 +631,7 @@
        return;
      }

      ReplicationDBCursor cursor = rsd.getCursorFrom(serverId, previousCN);
      ReplicaDBCursor cursor = rsd.getCursorFrom(serverId, previousCN);
      if (cursor != null)
      {
        try

 opends/src/server/org/opends/server/replication/server/ReplicationServerDomain.java

@@ -47,7 +47,7 @@
import org.opends.server.replication.common.*;
import org.opends.server.replication.protocol.*;
import org.opends.server.replication.server.changelog.api.ChangelogException;
import org.opends.server.replication.server.changelog.api.ReplicationDBCursor;
import org.opends.server.replication.server.changelog.api.ReplicaDBCursor;
import org.opends.server.replication.server.changelog.je.DbHandler;
import org.opends.server.types.*;

@@ -1263,17 +1263,17 @@

  /**
   * Creates and returns a cursor. When the cursor is not used anymore, the
   * caller MUST call the {@link ReplicationDBCursor#close()} method to free the
   * caller MUST call the {@link ReplicaDBCursor#close()} method to free the
   * resources and locks used by the cursor.
   *
   * @param serverId
   *          Identifier of the server for which the cursor is created.
   * @param startAfterCN
   *          Starting point for the cursor.
   * @return the created {@link ReplicationDBCursor}. Null when no DB is
   * @return the created {@link ReplicaDBCursor}. Null when no DB is
   *         available or the DB is empty for the provided serverId .
   */
  public ReplicationDBCursor getCursorFrom(int serverId,
  public ReplicaDBCursor getCursorFrom(int serverId,
      ChangeNumber startAfterCN)
  {
    DbHandler dbHandler = sourceDbHandlers.get(serverId);
@@ -1282,7 +1282,7 @@
      return null;
    }

    ReplicationDBCursor cursor;
    ReplicaDBCursor cursor;
    try
    {
      cursor = dbHandler.generateCursorFrom(startAfterCN);
@@ -2680,7 +2680,7 @@
          // to the Db and look for the change older than  eligible CN (cn14)
          if (eligibleCN.olderOrEqual(mostRecentDbCN)) {
            // let's try to seek the first change <= eligibleCN
            ReplicationDBCursor cursor = null;
            ReplicaDBCursor cursor = null;
            try {
              cursor = h.generateCursorFrom(eligibleCN);
              if (cursor != null && cursor.getChange() != null) {

 opends/src/server/org/opends/server/replication/server/changelog/api/ReplicaDBCursor.java

File was renamed from opends/src/server/org/opends/server/replication/server/changelog/api/ReplicationDBCursor.java
@@ -32,9 +32,9 @@

/**
 * This cursor allows to iterate through the changes received from a given
 * LDAP Server Identifier.
 * replica (Directory Server) in the topology.
 */
public interface ReplicationDBCursor extends Closeable
public interface ReplicaDBCursor extends Closeable
{

  /**
@@ -45,7 +45,7 @@
  UpdateMsg getChange();

  /**
   * Go to the next change in the ReplicationDB or in the server Queue.
   * Go to the next change in the ReplicaDB or in the server Queue.
   *
   * @return false if the cursor is already on the last change before this call.
   */

 opends/src/server/org/opends/server/replication/server/changelog/api/ReplicaDBCursorComparator.java

File was renamed from opends/src/server/org/opends/server/replication/server/changelog/api/ReplicationDBCursorComparator.java
@@ -33,14 +33,13 @@

/**
 * This class defines a {@link Comparator} that allows to know which
 * {@link ReplicationDBCursor} contain the next {@link UpdateMsg} in the order
 * {@link ReplicaDBCursor} contain the next {@link UpdateMsg} in the order
 * defined by the {@link ChangeNumber} of the {@link UpdateMsg}.
 */
public class ReplicationDBCursorComparator
              implements Comparator<ReplicationDBCursor>
public class ReplicaDBCursorComparator implements Comparator<ReplicaDBCursor>
{
  /**
   * Compare the {@link ChangeNumber} of the {@link ReplicationDBCursor}.
   * Compare the {@link ChangeNumber} of the {@link ReplicaDBCursor}.
   *
   * @param o1
   *          first cursor.
@@ -49,7 +48,7 @@
   * @return result of the comparison.
   */
  @Override
  public int compare(ReplicationDBCursor o1, ReplicationDBCursor o2)
  public int compare(ReplicaDBCursor o1, ReplicaDBCursor o2)
  {
    ChangeNumber csn1 = o1.getChange().getChangeNumber();
    ChangeNumber csn2 = o2.getChange().getChangeNumber();

 opends/src/server/org/opends/server/replication/server/changelog/je/DbHandler.java

@@ -43,7 +43,7 @@
import org.opends.server.replication.server.ReplicationServer;
import org.opends.server.replication.server.ReplicationServerDomain;
import org.opends.server.replication.server.changelog.api.ChangelogException;
import org.opends.server.replication.server.changelog.api.ReplicationDBCursor;
import org.opends.server.replication.server.changelog.api.ReplicaDBCursor;
import org.opends.server.replication.server.changelog.je.ReplicationDB.*;
import org.opends.server.types.Attribute;
import org.opends.server.types.Attributes;
@@ -59,7 +59,7 @@
 * server in the topology.
 * It is responsible for efficiently saving the updates that is received from
 * each master server into stable storage.
 * This class is also able to generate a {@link ReplicationDBCursor} that can be
 * This class is also able to generate a {@link ReplicaDBCursor} that can be
 * used to read all changes from a given {@link ChangeNumber}.
 *
 * This class publish some monitoring information below cn=monitor.
@@ -260,26 +260,26 @@
  }

  /**
   * Generate a new {@link ReplicationDBCursor} that allows to browse the db
   * Generate a new {@link ReplicaDBCursor} that allows to browse the db
   * managed by this dbHandler and starting at the position defined by a given
   * changeNumber.
   *
   * @param startAfterCN
   *          The position where the cursor must start.
   * @return a new {@link ReplicationDBCursor} that allows to browse the db
   * @return a new {@link ReplicaDBCursor} 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 ReplicationDBCursor generateCursorFrom(ChangeNumber startAfterCN)
  public ReplicaDBCursor generateCursorFrom(ChangeNumber startAfterCN)
      throws ChangelogException
  {
    if (startAfterCN == null)
    {
      flush();
    }
    return new JEReplicationDBCursor(db, startAfterCN, this);
    return new JEReplicaDBCursor(db, startAfterCN, this);
  }

  /**

 opends/src/server/org/opends/server/replication/server/changelog/je/JEReplicaDBCursor.java

File was renamed from opends/src/server/org/opends/server/replication/server/changelog/je/JEReplicationDBCursor.java
@@ -31,13 +31,13 @@
import org.opends.server.replication.common.ChangeNumber;
import org.opends.server.replication.protocol.UpdateMsg;
import org.opends.server.replication.server.changelog.api.ChangelogException;
import org.opends.server.replication.server.changelog.api.ReplicationDBCursor;
import org.opends.server.replication.server.changelog.api.ReplicaDBCursor;
import org.opends.server.replication.server.changelog.je.ReplicationDB.*;

/**
 * Berkeley DB JE implementation of {@link ReplicationDBCursor}.
 * Berkeley DB JE implementation of {@link ReplicaDBCursor}.
 */
public class JEReplicationDBCursor implements ReplicationDBCursor
public class JEReplicaDBCursor implements ReplicaDBCursor
{
  private UpdateMsg currentChange = null;
  private ReplServerDBCursor cursor = null;
@@ -46,8 +46,8 @@
  private ChangeNumber lastNonNullCurrentCN;

  /**
   * Creates a new JEReplicationDBCursor. All created cursor must be released by
   * the caller using the {@link #close()} method.
   * Creates a new {@link JEReplicaDBCursor}. All created cursor must be
   * released by the caller using the {@link #close()} method.
   *
   * @param db
   *          The db where the cursor must be created.
@@ -58,7 +58,7 @@
   * @throws ChangelogException
   *           if a database problem happened.
   */
  public JEReplicationDBCursor(ReplicationDB db, ChangeNumber startAfterCN,
  public JEReplicaDBCursor(ReplicationDB db, ChangeNumber startAfterCN,
      DbHandler dbHandler) throws ChangelogException
  {
    this.db = db;

 opends/tests/unit-tests-testng/src/server/org/opends/server/replication/server/changelog/je/DbHandlerTest.java

@@ -40,7 +40,7 @@
import org.opends.server.replication.protocol.DeleteMsg;
import org.opends.server.replication.server.ReplServerFakeConfiguration;
import org.opends.server.replication.server.ReplicationServer;
import org.opends.server.replication.server.changelog.api.ReplicationDBCursor;
import org.opends.server.replication.server.changelog.api.ReplicaDBCursor;
import org.opends.server.util.StaticUtils;
import org.testng.annotations.Test;

@@ -196,7 +196,7 @@
      return;
    }

        ReplicationDBCursor cursor = handler.generateCursorFrom(changeNumbers[0]);
        ReplicaDBCursor cursor = handler.generateCursorFrom(changeNumbers[0]);
    try
    {
      for (int i = 1; i < changeNumbers.length; i++)
@@ -217,7 +217,7 @@

  private void assertNotFound(DbHandler handler, ChangeNumber changeNumber)
  {
    ReplicationDBCursor cursor = null;
    ReplicaDBCursor cursor = null;
    try
    {
      cursor = handler.generateCursorFrom(changeNumber);