OPENDJ-1453 (CR-3938) Replica offline messages should be synced with updates

Jean-Noel Rouvignac

11.54.2014 aea0892feca2fd3d56c9c810debed6d22389454e

 opendj3-server-dev/src/server/org/opends/server/replication/server/ECLServerWriter.java

@@ -35,6 +35,7 @@
import org.opends.server.replication.protocol.DoneMsg;
import org.opends.server.replication.protocol.ECLUpdateMsg;
import org.opends.server.replication.protocol.Session;
import org.opends.server.replication.service.DSRSShutdownSync;
import org.opends.server.types.DirectoryException;
import org.opends.server.types.Entry;
import org.opends.server.workflowelement.externalchangelog.ECLSearchOperation;
@@ -47,7 +48,7 @@
 * This class defines a server writer, which is used to send changes to a
 * directory server.
 */
public class ECLServerWriter extends ServerWriter
class ECLServerWriter extends ServerWriter
{
  private static final LocalizedLogger logger = LocalizedLogger.getLoggerForThisClass();

@@ -56,7 +57,7 @@
  private final ReplicationServerDomain replicationServerDomain;
  private boolean suspended;
  private volatile boolean shutdown;
  private PersistentSearch mypsearch;
  private final PersistentSearch mypsearch;

  /**
   * Create a ServerWriter.
@@ -66,10 +67,10 @@
   * @param replicationServerDomain the ReplicationServerDomain of this
   *                    ServerWriter.
   */
  public ECLServerWriter(Session session, ECLServerHandler handler,
  ECLServerWriter(Session session, ECLServerHandler handler,
      ReplicationServerDomain replicationServerDomain)
  {
    super(session, handler, replicationServerDomain);
    super(session, handler, replicationServerDomain, new DSRSShutdownSync());

    setName("Replication ECL Writer Thread for operation " +
        handler.getOperationId());
@@ -79,21 +80,26 @@
    this.replicationServerDomain = replicationServerDomain;
    this.suspended = false;
    this.shutdown = false;
    this.mypsearch = findPersistentSearch(handler);
  }

    // Look for the psearch object related to this operation, the one that
    // will be notified with new entries to be returned.
    ECLWorkflowElement wfe =
        (ECLWorkflowElement) DirectoryServer
            .getWorkflowElement(ECLWorkflowElement.ECL_WORKFLOW_ELEMENT);
  /**
   * Look for the persistent search object related to this operation, the one
   * that will be notified with new entries to be returned.
   */
  private PersistentSearch findPersistentSearch(ECLServerHandler handler)
  {
    ECLWorkflowElement wfe = (ECLWorkflowElement)
        DirectoryServer.getWorkflowElement(ECLWorkflowElement.ECL_WORKFLOW_ELEMENT);
    for (PersistentSearch psearch : wfe.getPersistentSearches())
    {
      if (psearch.getSearchOperation().toString().equals(
          handler.getOperationId()))
      {
        mypsearch = psearch;
        break;
        return psearch;
      }
    }
    return null;
  }

  /**
@@ -101,7 +107,7 @@
   * waiting for the startCLSessionMsg. Then it may be
   * suspended between 2 jobs, each job being a separate search.
   */
  public synchronized void suspendWriter()
  private synchronized void suspendWriter()
  {
    suspended = true;
  }
@@ -109,7 +115,7 @@
  /**
   * Resume the writer.
   */
  public synchronized void resumeWriter()
  synchronized void resumeWriter()
  {
    suspended = false;
    notify();
@@ -180,7 +186,7 @@
   * @throws IOException when raised (connection closure)
   * @throws InterruptedException when raised
   */
  public void doIt() throws IOException, InterruptedException
  private void doIt() throws IOException, InterruptedException
  {
    while (true)
    {
@@ -230,7 +236,7 @@
  /**
   * Shutdown the writer.
   */
  public synchronized void shutdownWriter()
  synchronized void shutdownWriter()
  {
    shutdown = true;
    notify();