/*
 * 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
 * trunk/opends/resource/legal-notices/OpenDS.LICENSE
 * or https://OpenDS.dev.java.net/OpenDS.LICENSE.
 * 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
 * trunk/opends/resource/legal-notices/OpenDS.LICENSE.  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 2006-2009 Sun Microsystems, Inc.
 */
package org.opends.server.replication.server;

import org.opends.messages.*;

import static org.opends.server.loggers.ErrorLogger.logError;
import static org.opends.server.loggers.debug.DebugLogger.*;
import static org.opends.server.replication.common.StatusMachine.*;

import org.opends.server.loggers.debug.DebugTracer;
import static org.opends.messages.ReplicationMessages.*;
import static org.opends.server.util.StaticUtils.stackTraceToSingleLineString;

import java.io.IOException;
import java.util.Date;
import java.util.List;
import java.util.ArrayList;
import java.util.Map;
import java.util.Random;
import java.util.Set;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;

import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import org.opends.server.admin.std.server.MonitorProviderCfg;
import org.opends.server.api.MonitorProvider;
import org.opends.server.config.ConfigException;
import org.opends.server.core.DirectoryServer;
import org.opends.server.replication.common.AssuredMode;
import org.opends.server.replication.common.ChangeNumber;
import org.opends.server.replication.common.DSInfo;
import org.opends.server.replication.common.RSInfo;
import org.opends.server.replication.common.ServerState;
import org.opends.server.replication.common.ServerStatus;
import org.opends.server.replication.common.StatusMachine;
import org.opends.server.replication.common.StatusMachineEvent;
import org.opends.server.replication.protocol.*;
import org.opends.server.types.Attribute;
import org.opends.server.types.AttributeBuilder;
import org.opends.server.types.Attributes;
import org.opends.server.types.InitializationException;
import org.opends.server.util.TimeThread;

/**
 * This class defines a server handler, which handles all interaction with a
 * peer server (RS or DS).
 */
public class ServerHandler extends MonitorProvider<MonitorProviderCfg>
{

  /**
   * The tracer object for the debug logger.
   */
  private static final DebugTracer TRACER = getTracer();
  /**
   * Time during which the server will wait for existing thread to stop
   * during the shutdownWriter.
   */
  private static final int SHUTDOWN_JOIN_TIMEOUT = 30000;

  /*
   * Properties, filled if remote server is either a DS or a RS
   */
  private short serverId;
  private ProtocolSession session;
  private final MsgQueue msgQueue = new MsgQueue();
  private MsgQueue lateQueue = new MsgQueue();
  private ReplicationServerDomain replicationServerDomain = null;
  private String serverURL;

  // Number of update sent to the server
  private int outCount = 0;
  // Number of updates received from the server
  private int inCount = 0;

  // Number of updates received from the server in assured safe read mode
  private int assuredSrReceivedUpdates = 0;
  // Number of updates received from the server in assured safe read mode that
  // timed out
  private AtomicInteger assuredSrReceivedUpdatesTimeout = new AtomicInteger();
  // Number of updates sent to the server in assured safe read mode
  private int assuredSrSentUpdates = 0;
  // Number of updates sent to the server in assured safe read mode that timed
  // out
  private AtomicInteger assuredSrSentUpdatesTimeout = new AtomicInteger();
  // Number of updates received from the server in assured safe data mode
  private int assuredSdReceivedUpdates = 0;
  // Number of updates received from the server in assured safe data mode that
  // timed out
  private AtomicInteger assuredSdReceivedUpdatesTimeout = new AtomicInteger();
  // Number of updates sent to the server in assured safe data mode
  private int assuredSdSentUpdates = 0;
  // Number of updates sent to the server in assured safe data mode that timed
  // out
  private AtomicInteger assuredSdSentUpdatesTimeout = new AtomicInteger();

  private int maxReceiveQueue = 0;
  private int maxSendQueue = 0;
  private int maxReceiveDelay = 0;
  private int maxSendDelay = 0;
  private int maxQueueSize = 5000;
  private int maxQueueBytesSize = maxQueueSize * 100;
  private int restartReceiveQueue;
  private int restartSendQueue;
  private int restartReceiveDelay;
  private int restartSendDelay;
  private boolean serverIsLDAPserver;
  private boolean following = false;
  private ServerState serverState;
  private boolean activeWriter = true;
  private ServerWriter writer = null;
  private String baseDn = null;
  private int rcvWindow;
  private int rcvWindowSizeHalf;
  private int maxRcvWindow;
  private ServerReader reader;
  private Semaphore sendWindow;
  private int sendWindowSize;
  private boolean flowControl = false; // indicate that the server is
  // flow controlled and should
  // be stopped from sending messages.

  private int saturationCount = 0;
  private short replicationServerId;
  private short protocolVersion = -1;
  private long generationId = -1;

  // Group id of this remote server
  private byte groupId = (byte) -1;

  /*
   * Properties filled only if remote server is a DS
   */

  // Status of this DS (only used if this server handler represents a DS)
  private ServerStatus status = ServerStatus.INVALID_STATUS;
  // Referrals URLs this DS is exporting
  private List<String> refUrls = new ArrayList<String>();
  // Assured replication enabled on DS or not
  private boolean assuredFlag = false;
  // DS assured mode (relevant if assured replication enabled)
  private AssuredMode assuredMode = AssuredMode.SAFE_DATA_MODE;
  // DS safe data level (relevant if assured mode is safe data)
  private byte safeDataLevel = (byte) -1;

  /*
   * Properties filled only if remote server is a RS
   */
  private String serverAddressURL;
  /**
   * When this Handler is related to a remote replication server
   * this collection will contain as many elements as there are
   * LDAP servers connected to the remote replication server.
   */
  private final Map<Short, LightweightServerHandler> directoryServers =
    new ConcurrentHashMap<Short, LightweightServerHandler>();
  /**
   * The time in milliseconds between heartbeats from the replication
   * server.  Zero means heartbeats are off.
   */
  private long heartbeatInterval = 0;
  /**
   * The thread that will send heartbeats.
   */
  HeartbeatThread heartbeatThread = null;
  /**
   * Set when ServerWriter is stopping.
   */
  private boolean shutdownWriter = false;
  /**
   * Set when ServerHandler is stopping.
   */
  private AtomicBoolean shuttingDown = new AtomicBoolean(false);

  /**
   * Creates a new server handler instance with the provided socket.
   *
   * @param session The ProtocolSession used by the ServerHandler to
   *                 communicate with the remote entity.
   * @param queueSize The maximum number of update that will be kept
   *                  in memory by this ServerHandler.
   */
  public ServerHandler(ProtocolSession session, int queueSize)
  {
    super("Server Handler");
    this.session = session;
    this.maxQueueSize = queueSize;
    this.maxQueueBytesSize = queueSize * 100;
    this.protocolVersion = ProtocolVersion.getCurrentVersion();
  }

  /**
   * Creates a DSInfo structure representing this remote DS.
   * @return The DSInfo structure representing this remote DS
   */
  public DSInfo toDSInfo()
  {
    DSInfo dsInfo = new DSInfo(serverId, replicationServerId, generationId,
      status, assuredFlag, assuredMode, safeDataLevel, groupId, refUrls);

    return dsInfo;
  }

  /**
   * Creates a RSInfo structure representing this remote RS.
   * @return The RSInfo structure representing this remote RS
   */
  public RSInfo toRSInfo()
  {
    RSInfo rsInfo = new RSInfo(serverId, generationId, groupId);

    return rsInfo;
  }

  /**
   * Do the handshake with either the DS or RS and then create the reader and
   * writer thread.
   *
   * There are 2 possible handshake sequences: DS<->RS and RS<->RS. Each one are
   * divided into 2 logical consecutive phases (phase 1 and phase 2):
   *
   * DS<->RS (DS (always initiating connection) always sends first message):
   * -------
   *
   * phase 1:
   * DS --- ServerStartMsg ---> RS
   * DS <--- ReplServerStartMsg --- RS
   * phase 2:
   * DS --- StartSessionMsg ---> RS
   * DS <--- TopologyMsg --- RS
   *
   * RS<->RS (RS initiating connection always sends first message):
   * -------
   *
   * phase 1:
   * RS1 --- ReplServerStartMsg ---> RS2
   * RS1 <--- ReplServerStartMsg --- RS2
   * phase 2:
   * RS1 --- TopologyMsg ---> RS2
   * RS1 <--- TopologyMsg --- RS2
   *
   * @param baseDn baseDn of the ServerHandler when this is an outgoing conn.
   *               null if this is an incoming connection (listen).
   * @param replicationServerId The identifier of the replicationServer that
   *                            creates this server handler.
   * @param replicationServerURL The URL of the replicationServer that creates
   *                             this server handler.
   * @param windowSize the window size that this server handler must use.
   * @param sslEncryption For outgoing connections indicates whether encryption
   *                      should be used after the exchange of start messages.
   *                      Ignored for incoming connections.
   * @param replicationServer the ReplicationServer that created this server
   *                          handler.
   */
  public void start(String baseDn, short replicationServerId,
    String replicationServerURL,
    int windowSize, boolean sslEncryption,
    ReplicationServer replicationServer)
  {

    // The handshake phase must be done by blocking any access to structures
    // keeping info on connected servers, so that one can safely check for
    // pre-existence of a server, send a coherent snapshot of known topology
    // to peers, update the local view of the topology...
    //
    // For instance a kind of problem could be that while we connect with a
    // peer RS, a DS is connecting at the same time and we could publish the
    // connected DSs to the peer RS forgetting this last DS in the TopologyMsg.
    //
    // This method and every others that need to read/make changes to the
    // structures holding topology for the domain should:
    // - call ReplicationServerDomain.lock()
    // - read/modify structures
    // - call ReplicationServerDomain.release()
    //
    // More information is provided in comment of ReplicationServerDomain.lock()

    // If domain already exists, lock it until handshake is finished otherwise
    // it will be created and locked later in the method
    if (baseDn != null)
    {
      ReplicationServerDomain rsd =
        replicationServer.getReplicationServerDomain(baseDn, false);
      if (rsd != null)
      {
        try
        {
          rsd.lock();
        } catch (InterruptedException ex)
        {
          // Thread interrupted, return.
          return;
        }
      }
    }

    long oldGenerationId = -100;

    if (debugEnabled())
      TRACER.debugInfo("In " + replicationServer.getMonitorInstanceName() +
        " starts a new LS or RS " +
        ((baseDn == null) ? "incoming connection" : "outgoing connection"));

    this.replicationServerId = replicationServerId;
    rcvWindowSizeHalf = windowSize / 2;
    maxRcvWindow = windowSize;
    rcvWindow = windowSize;
    long localGenerationId = -1;
    ReplServerStartMsg outReplServerStartMsg = null;

    /**
     * This boolean prevents from logging a polluting error when connection\
     * aborted from a DS that wanted only to perform handshake phase 1 in order
     * to determine the best suitable RS:
     * 1) -> ServerStartMsg
     * 2) <- ReplServerStartMsg
     * 3) connection closure
     */
    boolean log_error_message = true;

    try
    {
      /*
       * PROCEDE WITH FIRST PHASE OF HANDSHAKE:
       * ServerStartMsg then ReplServerStartMsg (with a DS)
       * OR
       * ReplServerStartMsg then ReplServerStartMsg (with a RS)
       */

      if (baseDn != null) // Outgoing connection

      {
        // This is an outgoing connection. Publish our start message.
        this.baseDn = baseDn;

        // Get or create the ReplicationServerDomain
        replicationServerDomain =
          replicationServer.getReplicationServerDomain(baseDn, true);
        if (!replicationServerDomain.hasLock())
        {
          try
          {
            replicationServerDomain.lock();
          } catch (InterruptedException ex)
          {
            // Thread interrupted, return.
            return;
          }
        }
        localGenerationId = replicationServerDomain.getGenerationId();

        ServerState localServerState =
          replicationServerDomain.getDbServerState();
        outReplServerStartMsg = new ReplServerStartMsg(replicationServerId,
          replicationServerURL,
          baseDn, windowSize, localServerState,
          protocolVersion, localGenerationId,
          sslEncryption,
          replicationServer.getGroupId(),
          replicationServerDomain.
          getReplicationServer().getDegradedStatusThreshold());

        session.publish(outReplServerStartMsg);
      }

      // Wait and process ServerStartMsg or ReplServerStartMsg
      ReplicationMsg msg = session.receive();
      if (msg instanceof ServerStartMsg)
      {
        // The remote server is an LDAP Server.
        ServerStartMsg serverStartMsg = (ServerStartMsg) msg;

        generationId = serverStartMsg.getGenerationId();
        protocolVersion = ProtocolVersion.minWithCurrent(
          serverStartMsg.getVersion());
        serverId = serverStartMsg.getServerId();
        serverURL = serverStartMsg.getServerURL();
        this.baseDn = serverStartMsg.getBaseDn();
        this.serverState = serverStartMsg.getServerState();
        this.groupId = serverStartMsg.getGroupId();

        maxReceiveDelay = serverStartMsg.getMaxReceiveDelay();
        maxReceiveQueue = serverStartMsg.getMaxReceiveQueue();
        maxSendDelay = serverStartMsg.getMaxSendDelay();
        maxSendQueue = serverStartMsg.getMaxSendQueue();
        heartbeatInterval = serverStartMsg.getHeartbeatInterval();

        // The session initiator decides whether to use SSL.
        sslEncryption = serverStartMsg.getSSLEncryption();

        if (maxReceiveQueue > 0)
          restartReceiveQueue = (maxReceiveQueue > 1000 ? maxReceiveQueue -
            200 : maxReceiveQueue * 8 / 10);
        else
          restartReceiveQueue = 0;

        if (maxSendQueue > 0)
          restartSendQueue =
            (maxSendQueue > 1000 ? maxSendQueue - 200 : maxSendQueue * 8 /
            10);
        else
          restartSendQueue = 0;

        if (maxReceiveDelay > 0)
          restartReceiveDelay = (maxReceiveDelay > 10 ? maxReceiveDelay - 1
            : maxReceiveDelay);
        else
          restartReceiveDelay = 0;

        if (maxSendDelay > 0)
          restartSendDelay =
            (maxSendDelay > 10 ? maxSendDelay - 1 : maxSendDelay);
        else
          restartSendDelay = 0;

        if (heartbeatInterval < 0)
        {
          heartbeatInterval = 0;
        }

        serverIsLDAPserver = true;

        // Get or Create the ReplicationServerDomain
        replicationServerDomain =
          replicationServer.getReplicationServerDomain(this.baseDn, true);

        // Hack to be sure that if a server disconnects and reconnect, we
        // let the reader thread see the closure and cleanup any reference
        // to old connection
        replicationServerDomain.waitDisconnection(serverStartMsg.getServerId());

        if (!replicationServerDomain.hasLock())
        {
          try
          {
            replicationServerDomain.lock();
          } catch (InterruptedException ex)
          {
            // Thread interrupted, return.
            return;
          }
        }

        // Duplicate server ?
        if (!replicationServerDomain.checkForDuplicateDS(this))
        {
          closeSession(null);
          if ((replicationServerDomain != null) &&
            replicationServerDomain.hasLock())
            replicationServerDomain.release();
          return;
        }

        localGenerationId = replicationServerDomain.getGenerationId();

        ServerState localServerState =
          replicationServerDomain.getDbServerState();
        // This an incoming connection. Publish our start message
        ReplServerStartMsg replServerStartMsg =
          new ReplServerStartMsg(replicationServerId, replicationServerURL,
          this.baseDn, windowSize, localServerState,
          protocolVersion, localGenerationId,
          sslEncryption,
          replicationServer.getGroupId(),
          replicationServerDomain.
          getReplicationServer().getDegradedStatusThreshold());
        session.publish(replServerStartMsg);
        sendWindowSize = serverStartMsg.getWindowSize();

        /* Until here session is encrypted then it depends on the
        negotiation */
        if (!sslEncryption)
        {
          session.stopEncryption();
        }

        if (debugEnabled())
        {
          TRACER.debugInfo("In " +
            replicationServerDomain.getReplicationServer().
            getMonitorInstanceName() + ":" +
            "\nSH HANDSHAKE RECEIVED:\n" + serverStartMsg.toString() +
            "\nAND REPLIED:\n" + replServerStartMsg.toString());
        }
      } else if (msg instanceof ReplServerStartMsg)
      {
        // The remote server is a replication server
        ReplServerStartMsg inReplServerStartMsg = (ReplServerStartMsg) msg;
        protocolVersion = ProtocolVersion.minWithCurrent(
          inReplServerStartMsg.getVersion());
        generationId = inReplServerStartMsg.getGenerationId();
        serverId = inReplServerStartMsg.getServerId();
        serverURL = inReplServerStartMsg.getServerURL();
        int separator = serverURL.lastIndexOf(':');
        serverAddressURL =
          session.getRemoteAddress() + ":" + serverURL.substring(separator +
          1);
        serverIsLDAPserver = false;
        if (protocolVersion > ProtocolVersion.REPLICATION_PROTOCOL_V1)
        {
          // We support connection from a V1 RS
          // Only V2 protocol has the group id in repl server start message
          this.groupId = inReplServerStartMsg.getGroupId();
        }
        this.baseDn = inReplServerStartMsg.getBaseDn();

        if (baseDn == null) // Reply to incoming RS

        {
          // Get or create the ReplicationServerDomain
          replicationServerDomain =
            replicationServer.getReplicationServerDomain(this.baseDn, true);
          if (!replicationServerDomain.hasLock())
          {
            try
            {
              /**
               * Take the lock on the domain.
               * WARNING: Here we try to acquire the lock with a timeout. This
               * is for preventing a deadlock that may happen if there are cross
               * connection attempts (for same domain) from this replication
               * server and from a peer one:
               * Here is the scenario:
               * - RS1 connect thread takes the domain lock and starts
               * connection to RS2
               * - at the same time RS2 connect thread takes his domain lock and
               * start connection to RS2
               * - RS2 listen thread starts processing received
               * ReplServerStartMsg from RS1 and wants to acquire the lock on
               * the domain (here) but cannot as RS2 connect thread already has
               * it
               * - RS1 listen thread starts processing received
               * ReplServerStartMsg from RS2 and wants to acquire the lock on
               * the domain (here) but cannot as RS1 connect thread already has
               * it
               * => Deadlock: 4 threads are locked.
               * So to prevent that in such situation, the listen threads here
               * will both timeout trying to acquire the lock. The random time
               * for the timeout should allow on connection attempt to be
               * aborted whereas the other one should have time to finish in the
               * same time.
               * Warning: the minimum time (3s) should be big enough to allow
               * normal situation connections to terminate. The added random
               * time should represent a big enough range so that the chance to
               * have one listen thread timing out a lot before the peer one is
               * great. When the first listen thread times out, the remote
               * connect thread should release the lock and allow the peer
               * listen thread to take the lock it was waiting for and process
               * the connection attempt.
               */
              Random random = new Random();
              int randomTime = random.nextInt(6); // Random from 0 to 5
              // Wait at least 3 seconds + (0 to 5 seconds)
              long timeout = (long) (3000 + ( randomTime * 1000 ) );
              boolean noTimeout = replicationServerDomain.tryLock(timeout);
              if (!noTimeout)
              {
                // Timeout
                Message message = NOTE_TIMEOUT_WHEN_CROSS_CONNECTION.get(
                  this.baseDn,
                  Short.toString(serverId),
                  Short.toString(replicationServer.getServerId()));
                closeSession(message);
                return;
              }
            } catch (InterruptedException ex)
            {
              // Thread interrupted, return.
              return;
            }
          }
          localGenerationId = replicationServerDomain.getGenerationId();
          ServerState domServerState =
            replicationServerDomain.getDbServerState();

          // The session initiator decides whether to use SSL.
          sslEncryption = inReplServerStartMsg.getSSLEncryption();

          // Publish our start message
          outReplServerStartMsg = new ReplServerStartMsg(replicationServerId,
            replicationServerURL,
            this.baseDn, windowSize, domServerState,
            protocolVersion,
            localGenerationId,
            sslEncryption,
            replicationServer.getGroupId(),
            replicationServerDomain.
            getReplicationServer().getDegradedStatusThreshold());

          if (protocolVersion > ProtocolVersion.REPLICATION_PROTOCOL_V1)
          {
            session.publish(outReplServerStartMsg);
          } else {
            // We support connection from a V1 RS, send PDU with V1 form
            session.publish(outReplServerStartMsg,
              ProtocolVersion.REPLICATION_PROTOCOL_V1);
          }

          if (debugEnabled())
          {
            TRACER.debugInfo("In " +
              replicationServerDomain.getReplicationServer().
              getMonitorInstanceName() + ":" +
              "\nSH HANDSHAKE RECEIVED:\n" + inReplServerStartMsg.toString() +
              "\nAND REPLIED:\n" + outReplServerStartMsg.toString());
          }
        } else
        {
          // Did the remote RS answer with the DN we provided him ?
          if (!(this.baseDn.equals(baseDn)))
          {
            Message message = ERR_RS_DN_DOES_NOT_MATCH.get(
              this.baseDn.toString(),
              baseDn.toString());
            closeSession(message);
            if ((replicationServerDomain != null) &&
              replicationServerDomain.hasLock())
              replicationServerDomain.release();
            return;
          }

          if (debugEnabled())
          {
            TRACER.debugInfo("In " +
              replicationServerDomain.getReplicationServer().
              getMonitorInstanceName() + ":" +
              "\nSH HANDSHAKE SENT:\n" + outReplServerStartMsg.toString() +
              "\nAND RECEIVED:\n" + inReplServerStartMsg.toString());
          }
        }
        this.serverState = inReplServerStartMsg.getServerState();
        sendWindowSize = inReplServerStartMsg.getWindowSize();

        // Duplicate server ?
        if (!replicationServerDomain.checkForDuplicateRS(this))
        {
          closeSession(null);
          if ((replicationServerDomain != null) &&
            replicationServerDomain.hasLock())
            replicationServerDomain.release();
          return;
        }

        /* Until here session is encrypted then it depends on the
        negociation */
        if (!sslEncryption)
        {
          session.stopEncryption();
        }
      } else
      {
        // We did not recognize the message, close session as what
        // can happen after is undetermined and we do not want the server to
        // be disturbed
        closeSession(null);
        if ((replicationServerDomain != null) &&
          replicationServerDomain.hasLock())
          replicationServerDomain.release();
        return;
      }

      if (protocolVersion > ProtocolVersion.REPLICATION_PROTOCOL_V1)
      { // Only protocol version above V1 has a phase 2 handshake

        /*
         * NOW PROCEDE WITH SECOND PHASE OF HANDSHAKE:
         * TopologyMsg then TopologyMsg (with a RS)
         * OR
         * StartSessionMsg then TopologyMsg (with a DS)
         */

        TopologyMsg outTopoMsg = null;

        if (baseDn != null) // Outgoing connection to a RS

        {
          // Send our own TopologyMsg to remote RS
          outTopoMsg = replicationServerDomain.createTopologyMsgForRS();
          session.publish(outTopoMsg);
        }

        // Wait and process TopologyMsg or StartSessionMsg
        log_error_message = false;
        ReplicationMsg msg2 = session.receive();
        log_error_message = true;
        if (msg2 instanceof TopologyMsg)
        {
          // Remote RS sent his topo msg
          TopologyMsg inTopoMsg = (TopologyMsg) msg2;

          // CONNECTION WITH A RS

          // if the remote RS and the local RS have the same genID
          // then it's ok and nothing else to do
          if (generationId == localGenerationId)
          {
            if (debugEnabled())
            {
              TRACER.debugInfo("In " +
                replicationServerDomain.getReplicationServer().
                getMonitorInstanceName() + " RS with serverID=" + serverId +
                " is connected with the right generation ID");
            }
          } else
          {
            if (localGenerationId > 0)
            {
              // if the local RS is initialized
              if (generationId > 0)
              {
                // if the remote RS is initialized
                if (generationId != localGenerationId)
                {
                  // if the 2 RS have different generationID
                  if (replicationServerDomain.getGenerationIdSavedStatus())
                  {
                    // if the present RS has received changes regarding its
                    //     gen ID and so won't change without a reset
                    // then  we are just degrading the peer.
                    Message message = NOTE_BAD_GENERATION_ID_FROM_RS.get(
                      this.baseDn,
                      Short.toString(serverId),
                      Long.toString(generationId),
                      Long.toString(localGenerationId));
                    logError(message);
                  } else
                  {
                    // The present RS has never received changes regarding its
                    // gen ID.
                    //
                    // Example case:
                    // - we are in RS1
                    // - RS2 has genId2 from LS2 (genId2 <=> no data in LS2)
                    // - RS1 has genId1 from LS1 /genId1 comes from data in
                    //   suffix
                    // - we are in RS1 and we receive a START msg from RS2
                    // - Each RS keeps its genID / is degraded and when LS2
                    //   will be populated from LS1 everything will become ok.
                    //
                    // Issue:
                    // FIXME : Would it be a good idea in some cases to just
                    //         set the gen ID received from the peer RS
                    //         specially if the peer has a non null state and
                    //         we have a nul state ?
                    // replicationServerDomain.
                    // setGenerationId(generationId, false);
                    Message message = NOTE_BAD_GENERATION_ID_FROM_RS.get(
                      this.baseDn,
                      Short.toString(serverId),
                      Long.toString(generationId),
                      Long.toString(localGenerationId));
                    logError(message);
                  }
                }
              } else
              {
                // The remote RS has no genId. We don't change anything for the
                // current RS.
              }
            } else
            {
              // The local RS is not initialized - take the one received
              // WARNING: Must be done before computing topo message to send
              // to peer server as topo message must embed valid generation id
              // for our server
              oldGenerationId =
                replicationServerDomain.setGenerationId(generationId, false);
            }
          }

          if (baseDn == null) // Reply to the RS (incoming connection)

          {
            // Send our own TopologyMsg to remote RS
            outTopoMsg = replicationServerDomain.createTopologyMsgForRS();
            session.publish(outTopoMsg);

            if (debugEnabled())
            {
              TRACER.debugInfo("In " +
                replicationServerDomain.getReplicationServer().
                getMonitorInstanceName() + ":" +
                "\nSH HANDSHAKE RECEIVED:\n" + inTopoMsg.toString() +
                "\nAND REPLIED:\n" + outTopoMsg.toString());
            }
          } else
          {
            if (debugEnabled())
            {
              TRACER.debugInfo("In " +
                replicationServerDomain.getReplicationServer().
                getMonitorInstanceName() + ":" +
                "\nSH HANDSHAKE SENT:\n" + outTopoMsg.toString() +
                "\nAND RECEIVED:\n" + inTopoMsg.toString());
            }
          }

          // Alright, connected with new RS (either outgoing or incoming
          // connection): store handler.
          Map<Short, ServerHandler> connectedRSs =
            replicationServerDomain.getConnectedRSs();
          connectedRSs.put(serverId, this);

          // Process TopologyMsg sent by remote RS: store matching new info
          // (this will also warn our connected DSs of the new received info)
          replicationServerDomain.receiveTopoInfoFromRS(inTopoMsg, this, false);

        } else if (msg2 instanceof StartSessionMsg)
        {
          // CONNECTION WITH A DS

          // Process StartSessionMsg sent by remote DS
          StartSessionMsg startSessionMsg = (StartSessionMsg) msg2;

          this.status = startSessionMsg.getStatus();
          // Sanity check: is it a valid initial status?
          if (!isValidInitialStatus(this.status))
          {
            Message mesg = ERR_RS_INVALID_INIT_STATUS.get(
              this.status.toString(), this.baseDn.toString(),
              Short.toString(serverId));
            closeSession(mesg);
            if ((replicationServerDomain != null) &&
              replicationServerDomain.hasLock())
              replicationServerDomain.release();
            return;
          }
          this.refUrls = startSessionMsg.getReferralsURLs();
          this.assuredFlag = startSessionMsg.isAssured();
          this.assuredMode = startSessionMsg.getAssuredMode();
          this.safeDataLevel = startSessionMsg.getSafeDataLevel();

          /*
           * If we have already a generationID set for the domain
           * then
           *   if the connecting replica has not the same
           *   then it is degraded locally and notified by an error message
           * else
           *   we set the generationID from the one received
           *   (unsaved yet on disk . will be set with the 1rst change
           * received)
           */
          if (localGenerationId > 0)
          {
            if (generationId != localGenerationId)
            {
              Message message = NOTE_BAD_GENERATION_ID_FROM_DS.get(
                this.baseDn,
                Short.toString(serverId),
                Long.toString(generationId),
                Long.toString(localGenerationId));
              logError(message);
            }
          } else
          {
            // We are an empty Replicationserver
            if ((generationId > 0) && (!serverState.isEmpty()))
            {
              // If the LDAP server has already sent changes
              // it is not expected to connect to an empty RS
              Message message = NOTE_BAD_GENERATION_ID_FROM_DS.get(
                this.baseDn,
                Short.toString(serverId),
                Long.toString(generationId),
                Long.toString(localGenerationId));
              logError(message);
            } else
            {
              // The local RS is not initialized - take the one received
              // WARNING: Must be done before computing topo message to send
              // to peer server as topo message must embed valid generation id
              // for our server
              oldGenerationId =
                replicationServerDomain.setGenerationId(generationId, false);
            }
          }

          // Send our own TopologyMsg to DS
          outTopoMsg = replicationServerDomain.createTopologyMsgForDS(
            this.serverId);
          session.publish(outTopoMsg);

          if (debugEnabled())
          {
            TRACER.debugInfo("In " +
              replicationServerDomain.getReplicationServer().
              getMonitorInstanceName() + ":" +
              "\nSH HANDSHAKE RECEIVED:\n" + startSessionMsg.toString() +
              "\nAND REPLIED:\n" + outTopoMsg.toString());
          }

          // Alright, connected with new DS: store handler.
          Map<Short, ServerHandler> connectedDSs =
            replicationServerDomain.getConnectedDSs();
          connectedDSs.put(serverId, this);

          // Tell peer DSs a new DS just connected to us
          // No need to resend topo msg to this just new DS so not null
          // argument
          replicationServerDomain.sendTopoInfoToDSs(this);
          // Tell peer RSs a new DS just connected to us
          replicationServerDomain.sendTopoInfoToRSs();
        } else
        {
          // We did not recognize the message, close session as what
          // can happen after is undetermined and we do not want the server to
          // be disturbed
          closeSession(null);
          if ((replicationServerDomain != null) &&
            replicationServerDomain.hasLock())
            replicationServerDomain.release();
          return;
        }
      } else
      {
        // Terminate connection from a V1 RS

        // if the remote RS and the local RS have the same genID
        // then it's ok and nothing else to do
        if (generationId == localGenerationId)
        {
          if (debugEnabled())
          {
            TRACER.debugInfo("In " +
              replicationServerDomain.getReplicationServer().
              getMonitorInstanceName() + " RS V1 with serverID=" + serverId +
              " is connected with the right generation ID");
          }
        } else
        {
          if (localGenerationId > 0)
          {
            // if the local RS is initialized
            if (generationId > 0)
            {
              // if the remote RS is initialized
              if (generationId != localGenerationId)
              {
                // if the 2 RS have different generationID
                if (replicationServerDomain.getGenerationIdSavedStatus())
                {
                  // if the present RS has received changes regarding its
                  //     gen ID and so won't change without a reset
                  // then  we are just degrading the peer.
                  Message message = NOTE_BAD_GENERATION_ID_FROM_RS.get(
                    this.baseDn,
                    Short.toString(serverId),
                    Long.toString(generationId),
                    Long.toString(localGenerationId));
                  logError(message);
                } else
                {
                  // The present RS has never received changes regarding its
                  // gen ID.
                  //
                  // Example case:
                  // - we are in RS1
                  // - RS2 has genId2 from LS2 (genId2 <=> no data in LS2)
                  // - RS1 has genId1 from LS1 /genId1 comes from data in
                  //   suffix
                  // - we are in RS1 and we receive a START msg from RS2
                  // - Each RS keeps its genID / is degraded and when LS2
                  //   will be populated from LS1 everything will become ok.
                  //
                  // Issue:
                  // FIXME : Would it be a good idea in some cases to just
                  //         set the gen ID received from the peer RS
                  //         specially if the peer has a non null state and
                  //         we have a nul state ?
                  // replicationServerDomain.
                  // setGenerationId(generationId, false);
                  Message message = NOTE_BAD_GENERATION_ID_FROM_RS.get(
                    this.baseDn,
                    Short.toString(serverId),
                    Long.toString(generationId),
                    Long.toString(localGenerationId));
                  logError(message);
                }
              }
            } else
            {
              // The remote RS has no genId. We don't change anything for the
              // current RS.
            }
          } else
          {
            // The local RS is not initialized - take the one received
            oldGenerationId =
              replicationServerDomain.setGenerationId(generationId, false);
          }
        }

        // Alright, connected with new incoming V1 RS: store handler.
        Map<Short, ServerHandler> connectedRSs =
          replicationServerDomain.getConnectedRSs();
        connectedRSs.put(serverId, this);

        // Note: the supported scenario for V1->V2 upgrade is to upgrade 1 by 1
        // all the servers of the topology. We prefer not not send a TopologyMsg
        // for giving partial/false information to the V2 servers as for
        // instance we don't have the connected DS of the V1 RS...When the V1
        // RS will be upgraded in his turn, topo info will be sent and accurate.
        // That way, there is  no risk to have false/incomplete information in
        // other servers.
      }

      /*
       * FINALIZE INITIALIZATION:
       * CREATE READER AND WRITER, HEARTBEAT SYSTEM AND UPDATE MONITORING
       * SYSTEM
       */

      // Disable timeout for next communications
      session.setSoTimeout(0);
      // sendWindow MUST be created before starting the writer
      sendWindow = new Semaphore(sendWindowSize);

      writer = new ServerWriter(session, serverId,
        this, replicationServerDomain);
      reader = new ServerReader(session, serverId,
        this, replicationServerDomain);

      reader.start();
      writer.start();

      // Create a thread to send heartbeat messages.
      if (heartbeatInterval > 0)
      {
        heartbeatThread = new HeartbeatThread(
          "Replication Heartbeat to DS " + serverURL + " " + serverId +
          " for " + this.baseDn + " in RS " + replicationServerId,
          session, heartbeatInterval / 3);
        heartbeatThread.start();
      }

      // Create the status analyzer for the domain if not already started
      if (serverIsLDAPserver)
      {
        if (!replicationServerDomain.isRunningStatusAnalyzer())
        {
          if (debugEnabled())
            TRACER.debugInfo("In " + replicationServerDomain.
              getReplicationServer().
              getMonitorInstanceName() +
              " SH for remote server " + this.getMonitorInstanceName() +
              " is starting status analyzer");
          replicationServerDomain.startStatusAnalyzer();
        }
      }

      DirectoryServer.deregisterMonitorProvider(getMonitorInstanceName());
      DirectoryServer.registerMonitorProvider(this);
    } catch (NotSupportedOldVersionPDUException e)
    {
      // We do not need to support DS V1 connection, we just accept RS V1
      // connection:
      // We just trash the message, log the event for debug purpose and close
      // the connection
      if (debugEnabled())
      TRACER.debugInfo("In " + replicationServer.getMonitorInstanceName() + ":"
        + e.getMessage());
      closeSession(null);
    } catch (Exception e)
    {
      // We do not want polluting error log if error is due to normal session
      // aborted after handshake phase one from a DS that is searching for best
      // suitable RS.
      if ( log_error_message || (baseDn != null) )
      {
        // some problem happened, reject the connection
        MessageBuilder mb = new MessageBuilder();
        mb.append(ERR_REPLICATION_SERVER_CONNECTION_ERROR.get(
          this.getMonitorInstanceName()));
        mb.append(": " + stackTraceToSingleLineString(e));
        closeSession(mb.toMessage());
      } else
      {
        closeSession(null);
      }

      // If generation id of domain was changed, set it back to old value
      // We may have changed it as it was -1 and we received a value >0 from
      // peer server and the last topo message sent may have failed being
      // sent: in that case retrieve old value of generation id for
      // replication server domain
      if (oldGenerationId != -100)
      {
        replicationServerDomain.setGenerationId(oldGenerationId, false);
      }
    }

    // Release domain
    if ((replicationServerDomain != null) &&
      replicationServerDomain.hasLock())
      replicationServerDomain.release();
  }

  /*
   * Close the session logging the passed error message
   * Log nothing if message is null.
   */
  private void closeSession(Message msg)
  {
    if (msg != null)
    {
      logError(msg);
    }
    try
    {
      session.close();
    } catch (IOException ee)
    {
      // ignore
    }
  }

  /**
   * get the Server Id.
   *
   * @return the ID of the server to which this object is linked
   */
  public short getServerId()
  {
    return serverId;
  }

  /**
   * Retrieves the Address URL for this server handler.
   *
   * @return  The Address URL for this server handler,
   *          in the form of an IP address and port separated by a colon.
   */
  public String getServerAddressURL()
  {
    return serverAddressURL;
  }

  /**
   * Retrieves the URL for this server handler.
   *
   * @return  The URL for this server handler, in the form of an address and
   *          port separated by a colon.
   */
  public String getServerURL()
  {
    return serverURL;
  }

  /**
   * Increase the counter of updates sent to the server.
   */
  public void incrementOutCount()
  {
    outCount++;
  }

  /**
   * Increase the counter of update received from the server.
   */
  public void incrementInCount()
  {
    inCount++;
  }

  /**
   * Get the count of updates received from the server.
   * @return the count of update received from the server.
   */
  public int getInCount()
  {
    return inCount;
  }

  /**
   * Get the count of updates sent to this server.
   * @return  The count of update sent to this server.
   */
  public int getOutCount()
  {
    return outCount;
  }

  /**
   * Get the number of updates received from the server in assured safe read
   * mode.
   * @return The number of updates received from the server in assured safe read
   * mode
   */
  public int getAssuredSrReceivedUpdates()
  {
    return assuredSrReceivedUpdates;
  }

  /**
   * Get the number of updates received from the server in assured safe read
   * mode that timed out.
   * @return The number of updates received from the server in assured safe read
   * mode that timed out.
   */
  public AtomicInteger getAssuredSrReceivedUpdatesTimeout()
  {
    return assuredSrReceivedUpdatesTimeout;
  }

  /**
   * Get the number of updates sent to the server in assured safe read mode.
   * @return The number of updates sent to the server in assured safe read mode
   */
  public int getAssuredSrSentUpdates()
  {
    return assuredSrSentUpdates;
  }

  /**
   * Get the number of updates sent to the server in assured safe read mode that
   * timed out.
   * @return The number of updates sent to the server in assured safe read mode
   * that timed out.
   */
  public AtomicInteger getAssuredSrSentUpdatesTimeout()
  {
    return assuredSrSentUpdatesTimeout;
  }

    /**
   * Get the number of updates received from the server in assured safe data
   * mode.
   * @return The number of updates received from the server in assured safe data
   * mode
   */
  public int getAssuredSdReceivedUpdates()
  {
    return assuredSdReceivedUpdates;
  }

  /**
   * Get the number of updates received from the server in assured safe data
   * mode that timed out.
   * @return The number of updates received from the server in assured safe data
   * mode that timed out.
   */
  public AtomicInteger getAssuredSdReceivedUpdatesTimeout()
  {
    return assuredSdReceivedUpdatesTimeout;
  }

  /**
   * Get the number of updates sent to the server in assured safe data mode.
   * @return The number of updates sent to the server in assured safe data mode
   */
  public int getAssuredSdSentUpdates()
  {
    return assuredSdSentUpdates;
  }

  /**
   * Get the number of updates sent to the server in assured safe data mode that
   * timed out.
   * @return The number of updates sent to the server in assured safe data mode
   * that timed out.
   */
  public AtomicInteger getAssuredSdSentUpdatesTimeout()
  {
    return assuredSdSentUpdatesTimeout;
  }

  /**
   * Increment the number of updates received from the server in assured safe
   * read mode.
   */
  public void incrementAssuredSrReceivedUpdates()
  {
    assuredSrReceivedUpdates++;
  }

  /**
   * Increment the number of updates received from the server in assured safe
   * read mode that timed out.
   */
  public void incrementAssuredSrReceivedUpdatesTimeout()
  {
    assuredSrReceivedUpdatesTimeout.incrementAndGet();
  }

  /**
   * Increment the number of updates sent to the server in assured safe read
   * mode.
   */
  public void incrementAssuredSrSentUpdates()
  {
    assuredSrSentUpdates++;
  }

  /**
   * Increment the number of updates sent to the server in assured safe read
   * mode that timed out.
   */
  public void incrementAssuredSrSentUpdatesTimeout()
  {
    assuredSrSentUpdatesTimeout.incrementAndGet();
  }

  /**
   * Increment the number of updates received from the server in assured safe
   * data mode.
   */
  public void incrementAssuredSdReceivedUpdates()
  {
    assuredSdReceivedUpdates++;
  }

  /**
   * Increment the number of updates received from the server in assured safe
   * data mode that timed out.
   */
  public void incrementAssuredSdReceivedUpdatesTimeout()
  {
    assuredSdReceivedUpdatesTimeout.incrementAndGet();
  }

  /**
   * Increment the number of updates sent to the server in assured safe data
   * mode.
   */
  public void incrementAssuredSdSentUpdates()
  {
    assuredSdSentUpdates++;
  }

  /**
   * Increment the number of updates sent to the server in assured safe data
   * mode that timed out.
   */
  public void incrementAssuredSdSentUpdatesTimeout()
  {
    assuredSdSentUpdatesTimeout.incrementAndGet();
  }

  /**
   * Check is this server is saturated (this server has already been
   * sent a bunch of updates and has not processed them so they are staying
   * in the message queue for this server an the size of the queue
   * for this server is above the configured limit.
   *
   * The limit can be defined in number of updates or with a maximum delay
   *
   * @param changeNumber The changenumber to use to make the delay calculations.
   * @param sourceHandler The ServerHandler which is sending the update.
   * @return true is saturated false if not saturated.
   */
  public boolean isSaturated(ChangeNumber changeNumber,
    ServerHandler sourceHandler)
  {
    synchronized (msgQueue)
    {
      int size = msgQueue.count();

      if ((maxReceiveQueue > 0) && (size >= maxReceiveQueue))
        return true;

      if ((sourceHandler.maxSendQueue > 0) &&
        (size >= sourceHandler.maxSendQueue))
        return true;

      if (!msgQueue.isEmpty())
      {
        UpdateMsg firstUpdate = msgQueue.first();

        if (firstUpdate != null)
        {
          long timeDiff = changeNumber.getTimeSec() -
            firstUpdate.getChangeNumber().getTimeSec();

          if ((maxReceiveDelay > 0) && (timeDiff >= maxReceiveDelay))
            return true;

          if ((sourceHandler.maxSendDelay > 0) &&
            (timeDiff >= sourceHandler.maxSendDelay))
            return true;
        }
      }
      return false;
    }
  }

  /**
   * Check that the size of the Server Handler messages Queue has lowered
   * below the limit and therefore allowing the reception of messages
   * from other servers to restart.
   * @param source The ServerHandler which was sending the update.
   *        can be null.
   * @return true if the processing can restart
   */
  public boolean restartAfterSaturation(ServerHandler source)
  {
    synchronized (msgQueue)
    {
      int queueSize = msgQueue.count();
      if ((maxReceiveQueue > 0) && (queueSize >= restartReceiveQueue))
        return false;
      if ((source != null) && (source.maxSendQueue > 0) &&
        (queueSize >= source.restartSendQueue))
        return false;

      if (!msgQueue.isEmpty())
      {
        UpdateMsg firstUpdate = msgQueue.first();
        UpdateMsg lastUpdate = msgQueue.last();

        if ((firstUpdate != null) && (lastUpdate != null))
        {
          long timeDiff = lastUpdate.getChangeNumber().getTimeSec() -
            firstUpdate.getChangeNumber().getTimeSec();
          if ((maxReceiveDelay > 0) && (timeDiff >= restartReceiveDelay))
            return false;
          if ((source != null) && (source.maxSendDelay > 0) && (timeDiff >=
            source.restartSendDelay))
            return false;
        }
      }
    }
    return true;
  }

  /**
   * Check if the server associated to this ServerHandler is a replication
   * server.
   * @return true if the server associated to this ServerHandler is a
   *         replication server.
   */
  public boolean isReplicationServer()
  {
    return (!serverIsLDAPserver);
  }

  /**
   * Get the number of message in the receive message queue.
   * @return Size of the receive message queue.
   */
  public int getRcvMsgQueueSize()
  {
    synchronized (msgQueue)
    {
      /*
       * When the server is up to date or close to be up to date,
       * the number of updates to be sent is the size of the receive queue.
       */
      if (isFollowing())
        return msgQueue.count();
      else
      {
        /**
         * When the server  is not able to follow, the msgQueue
         * may become too large and therefore won't contain all the
         * changes. Some changes may only be stored in the backing DB
         * of the servers.
         * The total size of the receive queue is calculated by doing
         * the sum of the number of missing changes for every dbHandler.
         */
        ServerState dbState = replicationServerDomain.getDbServerState();
        return ServerState.diffChanges(dbState, serverState);
      }
    }
  }

  /**
   * Get an approximation of the delay by looking at the age of the oldest
   * message that has not been sent to this server.
   * This is an approximation because the age is calculated using the
   * clock of the server where the replicationServer is currently running
   * while it should be calculated using the clock of the server
   * that originally processed the change.
   *
   * The approximation error is therefore the time difference between
   *
   * @return the approximate delay for the connected server.
   */
  public long getApproxDelay()
  {
    long olderUpdateTime = getOlderUpdateTime();
    if (olderUpdateTime == 0)
      return 0;

    long currentTime = TimeThread.getTime();
    return ((currentTime - olderUpdateTime) / 1000);
  }

  /**
   * Get the age of the older change that has not yet been replicated
   * to the server handled by this ServerHandler.
   * @return The age if the older change has not yet been replicated
   *         to the server handled by this ServerHandler.
   */
  public Long getApproxFirstMissingDate()
  {
    Long result = (long) 0;

    // Get the older CN received
    ChangeNumber olderUpdateCN = getOlderUpdateCN();
    if (olderUpdateCN != null)
    {
      // If not present in the local RS db,
      // then approximate with the older update time
      result = olderUpdateCN.getTime();
    }
    return result;
  }

  /**
   * Get the older update time for that server.
   * @return The older update time.
   */
  public long getOlderUpdateTime()
  {
    ChangeNumber olderUpdateCN = getOlderUpdateCN();
    if (olderUpdateCN == null)
      return 0;
    return olderUpdateCN.getTime();
  }

  /**
   * Get the older Change Number for that server.
   * Returns null when the queue is empty.
   * @return The older change number.
   */
  public ChangeNumber getOlderUpdateCN()
  {
    ChangeNumber result = null;
    synchronized (msgQueue)
    {
      if (isFollowing())
      {
        if (msgQueue.isEmpty())
        {
          result = null;
        } else
        {
          UpdateMsg msg = msgQueue.first();
          result = msg.getChangeNumber();
        }
      } else
      {
        if (lateQueue.isEmpty())
        {
          // isFollowing is false AND lateQueue is empty
          // We may be at the very moment when the writer has emptyed the
          // lateQueue when it sent the last update. The writer will fill again
          // 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.

          ReplicationIteratorComparator comparator =
            new ReplicationIteratorComparator();
          SortedSet<ReplicationIterator> iteratorSortedSet =
            new TreeSet<ReplicationIterator>(comparator);
          try
          {
            // Build a list of candidates iterator (i.e. db i.e. server)
            for (short serverId : replicationServerDomain.getServers())
            {
              // get the last already sent CN from that server
              ChangeNumber lastCsn = serverState.getMaxChangeNumber(serverId);
              // get an iterator in this server db from that last change
              ReplicationIterator iterator =
                replicationServerDomain.getChangelogIterator(serverId, lastCsn);
              // if that iterator has changes, then it is a candidate
              // it is added in the sorted list at a position given by its
              // current change (see ReplicationIteratorComparator).
              if ((iterator != null) && (iterator.getChange() != null))
              {
                iteratorSortedSet.add(iterator);
              }
            }
            UpdateMsg msg = iteratorSortedSet.first().getChange();
            result = msg.getChangeNumber();
          } catch (Exception e)
          {
            result = null;
          } finally
          {
            for (ReplicationIterator iterator : iteratorSortedSet)
            {
              iterator.releaseCursor();
            }
          }
        } else
        {
          UpdateMsg msg = lateQueue.first();
          result = msg.getChangeNumber();
        }
      }
    }
    return result;
  }

  /**
   * Check if the LDAP server can follow the speed of the other servers.
   * @return true when the server has all the not yet sent changes
   *         in its queue.
   */
  public boolean isFollowing()
  {
    return following;
  }

  /**
   * Set the following flag of this server.
   * @param following the value that should be set.
   */
  public void setFollowing(boolean following)
  {
    this.following = following;
  }

  /**
   * Add an update to the list of updates that must be sent to the server
   * managed by this ServerHandler.
   *
   * @param update The update that must be added to the list of updates.
   * @param sourceHandler The server that sent the update.
   */
  public void add(UpdateMsg update, ServerHandler sourceHandler)
  {
    synchronized (msgQueue)
    {
      /*
       * If queue was empty the writer thread was probably asleep
       * waiting for some changes, wake it up
       */
      if (msgQueue.isEmpty())
        msgQueue.notify();

      msgQueue.add(update);

      /* TODO : size should be configurable
       * and larger than max-receive-queue-size
       */
      while ((msgQueue.count() > maxQueueSize) ||
          (msgQueue.bytesCount() > maxQueueBytesSize))
      {
        setFollowing(false);
        msgQueue.removeFirst();
      }
    }

    if (isSaturated(update.getChangeNumber(), sourceHandler))
    {
      sourceHandler.setSaturated(true);
    }

  }

  private void setSaturated(boolean value)
  {
    flowControl = value;
  }

  /**
   * Select the next update that must be sent to the server managed by this
   * ServerHandler.
   *
   * @return the next update that must be sent to the server managed by this
   *         ServerHandler.
   */
  public UpdateMsg take()
  {
    boolean interrupted = true;
    UpdateMsg msg = getnextMessage();

    /*
     * When we remove a message from the queue we need to check if another
     * server is waiting in flow control because this queue was too long.
     * This check might cause a performance penalty an therefore it
     * is not done for every message removed but only every few messages.
     */
    if (++saturationCount > 10)
    {
      saturationCount = 0;
      try
      {
        replicationServerDomain.checkAllSaturation();
      } catch (IOException e)
      {
      }
    }
    boolean acquired = false;
    do
    {
      try
      {
        acquired = sendWindow.tryAcquire((long) 500, TimeUnit.MILLISECONDS);
        interrupted = false;
      } catch (InterruptedException e)
      {
        // loop until not interrupted
      }
    } while (((interrupted) || (!acquired)) && (!shutdownWriter));
    if (msg != null)
    {
      incrementOutCount();
      if (msg.isAssured())
      {
        if (msg.getAssuredMode() == AssuredMode.SAFE_READ_MODE)
        {
          incrementAssuredSrSentUpdates();
        } else
        {
          if (!isLDAPserver())
            incrementAssuredSdSentUpdates();
        }
      }
    }
    return msg;
  }

  /**
   * Get the next update that must be sent to the server
   * from the message queue or from the database.
   *
   * @return The next update that must be sent to the server.
   */
  private UpdateMsg getnextMessage()
  {
    UpdateMsg msg;
    while (activeWriter == true)
    {
      if (following == false)
      {
        /* this server is late with regard to some other masters
         * in the topology or just joined the topology.
         * In such cases, we can't keep all changes in the queue
         * without saturating the memory, we therefore use
         * a lateQueue that is filled with a few changes from the changelogDB
         * If this server is able to close the gap, it will start using again
         * the regular msgQueue later.
         */
        if (lateQueue.isEmpty())
        {
          /*
           * Start from the server State
           * Loop until the queue high mark or until no more changes
           *   for each known LDAP master
           *      get the next CSN after this last one :
           *         - try to get next from the file
           *         - if not found in the file
           *             - try to get the next from the queue
           *   select the smallest of changes
           *   check if it is in the memory tree
           *     yes : lock memory tree.
           *           check all changes from the list, remove the ones that
           *           are already sent
           *           unlock memory tree
           *           restart as usual
           *   load this change on the delayList
           *
           */
          ReplicationIteratorComparator comparator =
            new ReplicationIteratorComparator();
          SortedSet<ReplicationIterator> iteratorSortedSet =
            new TreeSet<ReplicationIterator>(comparator);
          /* fill the lateQueue */
          for (short serverId : replicationServerDomain.getServers())
          {
            ChangeNumber lastCsn = serverState.getMaxChangeNumber(serverId);
            ReplicationIterator iterator =
              replicationServerDomain.getChangelogIterator(serverId, lastCsn);
            if (iterator != null)
            {
              if (iterator.getChange() != null)
              {
                iteratorSortedSet.add(iterator);
              } else
              {
                iterator.releaseCursor();
              }
            }
          }

          // The loop below relies on the fact that it is sorted based
          // on the currentChange of each iterator to consider the next
          // change across all servers.
          // Hence it is necessary to remove and eventual add again an iterator
          // when looping in order to keep consistent the order of the
          // iterators (see ReplicationIteratorComparator.
          while (!iteratorSortedSet.isEmpty() &&
                 (lateQueue.count()<100) &&
                 (lateQueue.bytesCount()<50000) )
          {
            ReplicationIterator iterator = iteratorSortedSet.first();
            iteratorSortedSet.remove(iterator);
            lateQueue.add(iterator.getChange());
            if (iterator.next())
              iteratorSortedSet.add(iterator);
            else
              iterator.releaseCursor();
          }
          for (ReplicationIterator iterator : iteratorSortedSet)
          {
            iterator.releaseCursor();
          }
          /*
           * Check if the first change in the lateQueue is also on the regular
           * queue
           */
          if (lateQueue.isEmpty())
          {
            synchronized (msgQueue)
            {
              if ((msgQueue.count() < maxQueueSize) &&
                  (msgQueue.bytesCount() < maxQueueBytesSize))
              {
                setFollowing(true);
              }
            }
          } else
          {
            msg = lateQueue.first();
            synchronized (msgQueue)
            {
              if (msgQueue.contains(msg))
              {
                /* we finally catch up with the regular queue */
                setFollowing(true);
                lateQueue.clear();
                UpdateMsg msg1;
                do
                {
                  msg1 = msgQueue.removeFirst();
                } while (!msg.getChangeNumber().equals(msg1.getChangeNumber()));
                this.updateServerState(msg);
                return msg;
              }
            }
          }
        } else
        {
          /* get the next change from the lateQueue */
          msg = lateQueue.removeFirst();
          this.updateServerState(msg);
          return msg;
        }
      }
      synchronized (msgQueue)
      {
        if (following == true)
        {
          try
          {
            while (msgQueue.isEmpty() && (following == true))
            {
              msgQueue.wait(500);
              if (!activeWriter)
                return null;
            }
          } catch (InterruptedException e)
          {
            return null;
          }
          if (following == true)
          {
            msg = msgQueue.removeFirst();
            if (this.updateServerState(msg))
            {
              /*
               * Only push the message if it has not yet been seen
               * by the other server.
               * Otherwise just loop to select the next message.
               */
              return msg;
            }
          }
        }
      }
    /*
     * Need to loop because following flag may have gone to false between
     * the first check at the beginning of this method
     * and the second check just above.
     */
    }
    return null;
  }

  /**
   * Update the serverState with the last message sent.
   *
   * @param msg the last update sent.
   * @return boolean indicating if the update was meaningful.
   */
  public boolean updateServerState(UpdateMsg msg)
  {
    return serverState.update(msg.getChangeNumber());
  }

  /**
   * Get the state of this server.
   *
   * @return ServerState the state for this server..
   */
  public ServerState getServerState()
  {
    return serverState;
  }

  /**
   * Sends an ack message to the server represented by this object.
   *
   * @param ack The ack message to be sent.
   * @throws IOException In case of Exception thrown sending the ack.
   */
  public void sendAck(AckMsg ack) throws IOException
  {
    session.publish(ack);
  }

  /**
   * Check type of server handled.
   *
   * @return true if the handled server is an LDAP server.
   *         false if the handled server is a replicationServer
   */
  public boolean isLDAPserver()
  {
    return serverIsLDAPserver;
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public void initializeMonitorProvider(MonitorProviderCfg configuration)
    throws ConfigException, InitializationException
  {
    // Nothing to do for now
  }

  /**
   * Retrieves the name of this monitor provider.  It should be unique among all
   * monitor providers, including all instances of the same monitor provider.
   *
   * @return  The name of this monitor provider.
   */
  @Override
  public String getMonitorInstanceName()
  {
    String str = serverURL + " " + String.valueOf(serverId);

    if (serverIsLDAPserver)
      return "Connected Replica " + str +
                ",cn=" + replicationServerDomain.getMonitorInstanceName();
    else
      return "Connected Replication Server " + str +
                ",cn=" + replicationServerDomain.getMonitorInstanceName();
  }

  /**
   * Retrieves the length of time in milliseconds that should elapse between
   * calls to the <CODE>updateMonitorData()</CODE> method.  A negative or zero
   * return value indicates that the <CODE>updateMonitorData()</CODE> method
   * should not be periodically invoked.
   *
   * @return  The length of time in milliseconds that should elapse between
   *          calls to the <CODE>updateMonitorData()</CODE> method.
   */
  @Override
  public long getUpdateInterval()
  {
    /* we don't wont to do polling on this monitor */
    return 0;
  }

  /**
   * Performs any processing periodic processing that may be desired to update
   * the information associated with this monitor.  Note that best-effort
   * attempts will be made to ensure that calls to this method come
   * <CODE>getUpdateInterval()</CODE> milliseconds apart, but no guarantees will
   * be made.
   */
  @Override
  public void updateMonitorData()
  {
    // As long as getUpdateInterval() returns 0, this will never get called
  }

  /**
   * Retrieves a set of attributes containing monitor data that should be
   * returned to the client if the corresponding monitor entry is requested.
   *
   * @return  A set of attributes containing monitor data that should be
   *          returned to the client if the corresponding monitor entry is
   *          requested.
   */
  @Override
  public ArrayList<Attribute> getMonitorData()
  {
    ArrayList<Attribute> attributes = new ArrayList<Attribute>();
    if (serverIsLDAPserver)
    {
      attributes.add(Attributes.create("replica", serverURL));
      attributes.add(Attributes.create("connected-to",
          this.replicationServerDomain.getReplicationServer()
              .getMonitorInstanceName()));

    }
    else
    {
      attributes.add(Attributes.create("Replication-Server",
          serverURL));
    }
    attributes.add(Attributes.create("server-id", String
        .valueOf(serverId)));
    attributes.add(Attributes.create("domain-name", baseDn.toString()));

    try
    {
      MonitorData md;
      md = replicationServerDomain.computeMonitorData();

      if (serverIsLDAPserver)
      {
        // Oldest missing update
        Long approxFirstMissingDate = md.getApproxFirstMissingDate(serverId);
        if ((approxFirstMissingDate != null) && (approxFirstMissingDate > 0))
        {
          Date date = new Date(approxFirstMissingDate);
          attributes.add(Attributes.create(
              "approx-older-change-not-synchronized", date.toString()));
          attributes.add(Attributes.create(
              "approx-older-change-not-synchronized-millis", String
              .valueOf(approxFirstMissingDate)));
        }

        // Missing changes
        long missingChanges = md.getMissingChanges(serverId);
        attributes.add(Attributes.create("missing-changes", String
            .valueOf(missingChanges)));

        // Replication delay
        long delay = md.getApproxDelay(serverId);
        attributes.add(Attributes.create("approximate-delay", String
            .valueOf(delay)));

        /* get the Server State */
        AttributeBuilder builder = new AttributeBuilder("server-state");
        ServerState state = md.getLDAPServerState(serverId);
        if (state != null)
        {
          for (String str : state.toStringSet())
          {
            builder.add(str);
          }
          attributes.add(builder.toAttribute());
        }
      }
      else
      {
        // Missing changes
        long missingChanges = md.getMissingChangesRS(serverId);
        attributes.add(Attributes.create("missing-changes", String
            .valueOf(missingChanges)));

        /* get the Server State */
        AttributeBuilder builder = new AttributeBuilder("server-state");
        ServerState state = md.getRSStates(serverId);
        if (state != null)
        {
          for (String str : state.toStringSet())
          {
            builder.add(str);
          }
          attributes.add(builder.toAttribute());
        }
      }
    }
    catch (Exception e)
    {
      Message message =
        ERR_ERROR_RETRIEVING_MONITOR_DATA.get(stackTraceToSingleLineString(e));
      // We failed retrieving the monitor data.
      attributes.add(Attributes.create("error", message.toString()));
    }

    attributes.add(
        Attributes.create("queue-size", String.valueOf(msgQueue.count())));
    attributes.add(
        Attributes.create(
            "queue-size-bytes", String.valueOf(msgQueue.bytesCount())));
    attributes.add(
        Attributes.create(
            "following", String.valueOf(following)));

    // Deprecated
    attributes.add(Attributes.create("max-waiting-changes", String
        .valueOf(maxQueueSize)));
    attributes.add(Attributes.create("sent-updates", String
        .valueOf(getOutCount())));
    attributes.add(Attributes.create("received-updates", String
        .valueOf(getInCount())));

    // Assured counters
    attributes.add(Attributes.create("assured-sr-received-updates", String
        .valueOf(getAssuredSrReceivedUpdates())));
    attributes.add(Attributes.create("assured-sr-received-updates-timeout",
      String .valueOf(getAssuredSrReceivedUpdatesTimeout())));
    attributes.add(Attributes.create("assured-sr-sent-updates", String
        .valueOf(getAssuredSrSentUpdates())));
    attributes.add(Attributes.create("assured-sr-sent-updates-timeout", String
        .valueOf(getAssuredSrSentUpdatesTimeout())));
    attributes.add(Attributes.create("assured-sd-received-updates", String
        .valueOf(getAssuredSdReceivedUpdates())));
    if (!isLDAPserver())
    {
      attributes.add(Attributes.create("assured-sd-sent-updates",
        String.valueOf(getAssuredSdSentUpdates())));
      attributes.add(Attributes.create("assured-sd-sent-updates-timeout",
        String.valueOf(getAssuredSdSentUpdatesTimeout())));
    } else
    {
      attributes.add(Attributes.create("assured-sd-received-updates-timeout",
        String.valueOf(getAssuredSdReceivedUpdatesTimeout())));
    }

    // Window stats
    attributes.add(Attributes.create("max-send-window", String
        .valueOf(sendWindowSize)));
    attributes.add(Attributes.create("current-send-window", String
        .valueOf(sendWindow.availablePermits())));
    attributes.add(Attributes.create("max-rcv-window", String
        .valueOf(maxRcvWindow)));
    attributes.add(Attributes.create("current-rcv-window", String
        .valueOf(rcvWindow)));

    // Encryption
    attributes.add(Attributes.create("ssl-encryption", String
        .valueOf(session.isEncrypted())));

    // Data generation
    attributes.add(Attributes.create("generation-id", String
        .valueOf(generationId)));

    return attributes;
  }

  /**
   * Shutdown This ServerHandler.
   */
  public void shutdown()
  {
    /*
     * Shutdown ServerWriter
     */
    shutdownWriter = true;
    activeWriter = false;
    synchronized (msgQueue)
    {
      /* wake up the writer thread on an empty queue so that it disappear */
      msgQueue.clear();
      msgQueue.notify();
      msgQueue.notifyAll();
    }

    /*
     * Close session to end ServerReader or ServerWriter
     */
    try
    {
      session.close();
    } catch (IOException e)
    {
      // ignore.
    }

    /*
     * Stop the remote LSHandler
     */
    synchronized (directoryServers)
    {
      for (LightweightServerHandler lsh : directoryServers.values())
      {
        lsh.stopHandler();
      }
      directoryServers.clear();
    }

    /*
     * Stop the heartbeat thread.
     */
    if (heartbeatThread != null)
    {
      heartbeatThread.shutdown();
    }

    DirectoryServer.deregisterMonitorProvider(getMonitorInstanceName());

    /*
     * Be sure to wait for ServerWriter and ServerReader death
     * It does not matter if we try to stop a thread which is us (reader
     * or writer), but we must not wait for our own thread death.
     */
    try
    {
      if ((writer != null) && (!(Thread.currentThread().equals(writer))))
      {

        writer.join(SHUTDOWN_JOIN_TIMEOUT);
      }
      if ((reader != null) && (!(Thread.currentThread().equals(reader))))
      {
        reader.join(SHUTDOWN_JOIN_TIMEOUT);
      }
    } catch (InterruptedException e)
    {
      // don't try anymore to join and return.
    }
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public String toString()
  {
    String localString;
    if (serverId != 0)
    {
      if (serverIsLDAPserver)
        localString = "Directory Server ";
      else
        localString = "Replication Server ";


      localString += serverId + " " + serverURL + " " + baseDn;
    } else
      localString = "Unknown server";

    return localString;
  }

  /**
   * Decrement the protocol window, then check if it is necessary
   * to send a WindowMsg and send it.
   *
   * @throws IOException when the session becomes unavailable.
   */
  public synchronized void decAndCheckWindow() throws IOException
  {
    rcvWindow--;
    checkWindow();
  }

  /**
   * Check the protocol window and send WindowMsg if necessary.
   *
   * @throws IOException when the session becomes unavailable.
   */
  public synchronized void checkWindow() throws IOException
  {
    if (rcvWindow < rcvWindowSizeHalf)
    {
      if (flowControl)
      {
        if (replicationServerDomain.restartAfterSaturation(this))
        {
          flowControl = false;
        }
      }
      if (!flowControl)
      {
        WindowMsg msg = new WindowMsg(rcvWindowSizeHalf);
        session.publish(msg);
        rcvWindow += rcvWindowSizeHalf;
      }
    }
  }

  /**
   * Update the send window size based on the credit specified in the
   * given window message.
   *
   * @param windowMsg The Window Message containing the information
   *                  necessary for updating the window size.
   */
  public void updateWindow(WindowMsg windowMsg)
  {
    sendWindow.release(windowMsg.getNumAck());
  }

  /**
   * Get our heartbeat interval.
   * @return Our heartbeat interval.
   */
  public long getHeartbeatInterval()
  {
    return heartbeatInterval;
  }

  /**
   * Processes a routable message.
   *
   * @param msg The message to be processed.
   */
  public void process(RoutableMsg msg)
  {
    if (debugEnabled())
      TRACER.debugInfo("In " + replicationServerDomain.getReplicationServer().
        getMonitorInstanceName() +
        " SH for remote server " + this.getMonitorInstanceName() + ":" +
        "\nprocesses received msg:\n" + msg);
    replicationServerDomain.process(msg, this);
  }

  /**
   * Sends the provided TopologyMsg to the peer server.
   *
   * @param topoMsg The TopologyMsg message to be sent.
   * @throws IOException When it occurs while sending the message,
   *
   */
  public void sendTopoInfo(TopologyMsg topoMsg)
    throws IOException
  {
    // V1 Rs do not support the TopologyMsg
    if (protocolVersion > ProtocolVersion.REPLICATION_PROTOCOL_V1)
    {
      if (debugEnabled())
        TRACER.debugInfo("In " + replicationServerDomain.getReplicationServer().
          getMonitorInstanceName() +
          " SH for remote server " + this.getMonitorInstanceName() + ":" +
          "\nsends message:\n" + topoMsg);

      session.publish(topoMsg);
    }
  }

  /**
   * Stores topology information received from a peer RS and that must be kept
   * in RS handler.
   *
   * @param topoMsg The received topology message
   */
  public void receiveTopoInfoFromRS(TopologyMsg topoMsg)
  {
    // Store info for remote RS
    List<RSInfo> rsInfos = topoMsg.getRsList();
    // List should only contain RS info for sender
    RSInfo rsInfo = rsInfos.get(0);
    generationId = rsInfo.getGenerationId();
    groupId = rsInfo.getGroupId();

    /**
     * Store info for DSs connected to the peer RS
     */
    List<DSInfo> dsInfos = topoMsg.getDsList();

    synchronized (directoryServers)
    {
      // Removes the existing structures
      for (LightweightServerHandler lsh : directoryServers.values())
      {
        lsh.stopHandler();
      }
      directoryServers.clear();

      // Creates the new structure according to the message received.
      for (DSInfo dsInfo : dsInfos)
      {
        LightweightServerHandler lsh = new LightweightServerHandler(this,
            serverId, dsInfo.getDsId(), dsInfo.getGenerationId(),
            dsInfo.getGroupId(), dsInfo.getStatus(), dsInfo.getRefUrls(),
            dsInfo.isAssured(), dsInfo.getAssuredMode(),
            dsInfo.getSafeDataLevel());
        lsh.startHandler();
        directoryServers.put(lsh.getServerId(), lsh);
      }
    }
  }

  /**
   * Process message of a remote server changing his status.
   * @param csMsg The message containing the new status
   * @return The new server status of the DS
   */
  public ServerStatus processNewStatus(ChangeStatusMsg csMsg)
  {

    // Sanity check
    if (!serverIsLDAPserver)
    {
      Message msg =
        ERR_RECEIVED_CHANGE_STATUS_NOT_FROM_DS.get(baseDn.toString(),
        Short.toString(serverId), csMsg.toString());
      logError(msg);
      return ServerStatus.INVALID_STATUS;
    }

    // Get the status the DS just entered
    ServerStatus reqStatus = csMsg.getNewStatus();
    // Translate new status to a state machine event
    StatusMachineEvent event = StatusMachineEvent.statusToEvent(reqStatus);
    if (event == StatusMachineEvent.INVALID_EVENT)
    {
      Message msg = ERR_RS_INVALID_NEW_STATUS.get(reqStatus.toString(),
        baseDn.toString(), Short.toString(serverId));
      logError(msg);
      return ServerStatus.INVALID_STATUS;
    }

    // Check state machine allows this new status
    ServerStatus newStatus = StatusMachine.computeNewStatus(status, event);
    if (newStatus == ServerStatus.INVALID_STATUS)
    {
      Message msg = ERR_RS_CANNOT_CHANGE_STATUS.get(baseDn.toString(),
        Short.toString(serverId), status.toString(), event.toString());
      logError(msg);
      return ServerStatus.INVALID_STATUS;
    }

    status = newStatus;

    return status;
  }

  /**
   * Change the status according to the event generated from the status
   * analyzer.
   * @param event The event to be used for new status computation
   * @return The new status of the DS
   * @throws IOException When raised by the underlying session
   */
  public ServerStatus changeStatusFromStatusAnalyzer(StatusMachineEvent event)
    throws IOException
  {
    // Check state machine allows this new status (Sanity check)
    ServerStatus newStatus = StatusMachine.computeNewStatus(status, event);
    if (newStatus == ServerStatus.INVALID_STATUS)
    {
      Message msg = ERR_RS_CANNOT_CHANGE_STATUS.get(baseDn.toString(),
        Short.toString(serverId), status.toString(), event.toString());
      logError(msg);
      // Status analyzer must only change from NORMAL_STATUS to DEGRADED_STATUS
      // and vice versa. We may are being trying to change the status while for
      // instance another status has just been entered: e.g a full update has
      // just been engaged. In that case, just ignore attempt to change the
      // status
      return newStatus;
    }

    // Send message requesting to change the DS status
    ChangeStatusMsg csMsg = new ChangeStatusMsg(newStatus,
      ServerStatus.INVALID_STATUS);

    if (debugEnabled())
    {
      TRACER.debugInfo(
        "In RS " +
        replicationServerDomain.getReplicationServer().getServerId() +
        " Sending change status from status analyzer to " + getServerId() +
        " for baseDn " + baseDn + ":\n" + csMsg);
    }

    session.publish(csMsg);

    status = newStatus;

    return newStatus;
  }

  /**
   * When this handler is connected to a replication server, specifies if
   * a wanted server is connected to this replication server.
   *
   * @param wantedServer The server we want to know if it is connected
   * to the replication server represented by this handler.
   * @return boolean True is the wanted server is connected to the server
   * represented by this handler.
   */
  public boolean isRemoteLDAPServer(short wantedServer)
  {
    synchronized (directoryServers)
    {
      for (LightweightServerHandler server : directoryServers.values())
      {
        if (wantedServer == server.getServerId())
        {
          return true;
        }
      }
      return false;
    }
  }

  /**
   * When the handler is connected to a replication server, specifies the
   * replication server has remote LDAP servers connected to it.
   *
   * @return boolean True is the replication server has remote LDAP servers
   * connected to it.
   */
  public boolean hasRemoteLDAPServers()
  {
    synchronized (directoryServers)
    {
      return !directoryServers.isEmpty();
    }
  }

  /**
   * Send an InitializeRequestMessage to the server connected through this
   * handler.
   *
   * @param msg The message to be processed
   * @throws IOException when raised by the underlying session
   */
  public void send(RoutableMsg msg) throws IOException
  {
    if (debugEnabled())
      TRACER.debugInfo("In " +
        replicationServerDomain.getReplicationServer().
        getMonitorInstanceName() +
        " SH for remote server " + this.getMonitorInstanceName() + ":" +
        "\nsends message:\n" + msg);
    session.publish(msg);
  }

  /**
   * Send an ErrorMsg to the peer.
   *
   * @param errorMsg The message to be sent
   * @throws IOException when raised by the underlying session
   */
  public void sendError(ErrorMsg errorMsg) throws IOException
  {
    session.publish(errorMsg);
  }

  /**
   * Process the reception of a WindowProbeMsg message.
   *
   * @param  windowProbeMsg The message to process.
   *
   * @throws IOException    When the session becomes unavailable.
   */
  public void process(WindowProbeMsg windowProbeMsg) throws IOException
  {
    if (rcvWindow > 0)
    {
      // The LDAP server believes that its window is closed
      // while it is not, this means that some problem happened in the
      // window exchange procedure !
      // lets update the LDAP server with out current window size and hope
      // that everything will work better in the futur.
      // TODO also log an error message.
      WindowMsg msg = new WindowMsg(rcvWindow);
      session.publish(msg);
    } else
    {
      // Both the LDAP server and the replication server believes that the
      // window is closed. Lets check the flowcontrol in case we
      // can now resume operations and send a windowMessage if necessary.
      checkWindow();
    }
  }

  /**
   * Returns the value of generationId for that handler.
   * @return The value of the generationId.
   */
  public long getGenerationId()
  {
    return generationId;
  }

  /**
   * Sends a message containing a generationId to a peer server.
   * The peer is expected to be a replication server.
   *
   * @param  msg         The GenerationIdMessage message to be sent.
   * @throws IOException When it occurs while sending the message,
   *
   */
  public void forwardGenerationIdToRS(ResetGenerationIdMsg msg)
    throws IOException
  {
    session.publish(msg);
  }

  /**
   * Set a new generation ID.
   *
   * @param generationId The new generation ID
   *
   */
  public void setGenerationId(long generationId)
  {
    this.generationId = generationId;
  }

  /**
   * Returns the Replication Server Domain to which belongs this server handler.
   *
   * @return The replication server domain.
   */
  public ReplicationServerDomain getDomain()
  {
    return this.replicationServerDomain;
  }

  /**
   * Return a Set containing the servers known by this replicationServer.
   * @return a set containing the servers known by this replicationServer.
   */
  public Set<Short> getConnectedDirectoryServerIds()
  {
    synchronized (directoryServers)
    {
      return directoryServers.keySet();
    }
  }

  /**
   * Order the peer DS server to change his status or close the connection
   * according to the requested new generation id.
   * @param newGenId The new generation id to take into account
   * @throws IOException If IO error occurred.
   */
  public void changeStatusForResetGenId(long newGenId)
    throws IOException
  {
    StatusMachineEvent event = null;

    if (newGenId == -1)
    {
      // The generation id is being made invalid, let's put the DS
      // into BAD_GEN_ID_STATUS
      event = StatusMachineEvent.TO_BAD_GEN_ID_STATUS_EVENT;
    } else
    {
      if (newGenId == generationId)
      {
        if (status == ServerStatus.BAD_GEN_ID_STATUS)
        {
          // This server has the good new reference generation id.
          // Close connection with him to force his reconnection: DS will
          // reconnect in NORMAL_STATUS or DEGRADED_STATUS.

          if (debugEnabled())
          {
            TRACER.debugInfo(
              "In RS " +
              replicationServerDomain.getReplicationServer().getServerId() +
              ". Closing connection to DS " + getServerId() +
              " for baseDn " + baseDn + " to force reconnection as new local" +
              " generation id and remote one match and DS is in bad gen id: " +
              newGenId);
          }

          // Connection closure must not be done calling RSD.stopHandler() as it
          // would rewait the RSD lock that we already must have entering this
          // method. This would lead to a reentrant lock which we do not want.
          // So simply close the session, this will make the hang up appear
          // after the reader thread that took the RSD lock realeases it.
          try
          {
            if (session != null)
              session.close();
          } catch (IOException e)
          {
            // ignore
          }

          // NOT_CONNECTED_STATUS is the last one in RS session life: handler
          // will soon disappear after this method call...
          status = ServerStatus.NOT_CONNECTED_STATUS;
          return;
        } else
        {
          if (debugEnabled())
          {
            TRACER.debugInfo(
              "In RS " +
              replicationServerDomain.getReplicationServer().getServerId() +
              ". DS " + getServerId() + " for baseDn " + baseDn +
              " has already generation id " + newGenId +
              " so no ChangeStatusMsg sent to him.");
          }
          return;
        }
      } else
      {
        // This server has a bad generation id compared to new reference one,
        // let's put it into BAD_GEN_ID_STATUS
        event = StatusMachineEvent.TO_BAD_GEN_ID_STATUS_EVENT;
      }
    }

    if ((event == StatusMachineEvent.TO_BAD_GEN_ID_STATUS_EVENT) &&
      (status == ServerStatus.FULL_UPDATE_STATUS))
    {
      // Prevent useless error message (full update status cannot lead to bad
      // gen status)
      Message message = NOTE_BAD_GEN_ID_IN_FULL_UPDATE.get(
        Short.toString(replicationServerDomain.
        getReplicationServer().getServerId()),
        baseDn.toString(),
        Short.toString(serverId),
        Long.toString(generationId),
        Long.toString(newGenId));
      logError(message);
      return;
    }

    ServerStatus newStatus = StatusMachine.computeNewStatus(status, event);

    if (newStatus == ServerStatus.INVALID_STATUS)
    {
      Message msg = ERR_RS_CANNOT_CHANGE_STATUS.get(baseDn.toString(),
        Short.toString(serverId), status.toString(), event.toString());
      logError(msg);
      return;
    }

    // Send message requesting to change the DS status
    ChangeStatusMsg csMsg = new ChangeStatusMsg(newStatus,
      ServerStatus.INVALID_STATUS);

    if (debugEnabled())
    {
      TRACER.debugInfo(
        "In RS " +
        replicationServerDomain.getReplicationServer().getServerId() +
        " Sending change status for reset gen id to " + getServerId() +
        " for baseDn " + baseDn + ":\n" + csMsg);
    }

    session.publish(csMsg);

    status = newStatus;
  }

  /**
   * Set the shut down flag to true and returns the previous value of the flag.
   * @return The previous value of the shut down flag
   */
  public boolean engageShutdown()
  {
    // Use thread safe boolean
    return shuttingDown.getAndSet(true);
  }

  /**
   * Gets the status of the connected DS.
   * @return The status of the connected DS.
   */
  public ServerStatus getStatus()
  {
    return status;
  }

  /**
   * Gets the protocol version used with this remote server.
   * @return The protocol version used with this remote server.
   */
  public short getProtocolVersion()
  {
    return protocolVersion;
  }

  /**
   * Add the DSinfos of the connected Directory Servers
   * to the List of DSInfo provided as a parameter.
   *
   * @param dsInfos The List of DSInfo that should be updated
   *                with the DSInfo for the directoryServers
   *                connected to this ServerHandler.
   */
  public void addDSInfos(List<DSInfo> dsInfos)
  {
    synchronized (directoryServers)
    {
      for (LightweightServerHandler ls : directoryServers.values())
      {
        dsInfos.add(ls.toDSInfo());
      }
    }
  }

  /**
   * Gets the group id of the server represented by this object.
   * @return The group id of the server represented by this object.
   */
  public byte getGroupId()
  {
    return groupId;
  }
}
