/*
 * The contents of this file are subject to the terms of the Common Development and
 * Distribution License (the License). You may not use this file except in compliance with the
 * License.
 *
 * You can obtain a copy of the License at legal/CDDLv1.0.txt. See the License for the
 * specific language governing permission and limitations under the License.
 *
 * When distributing Covered Software, include this CDDL Header Notice in each file and include
 * the License file at legal/CDDLv1.0.txt. If applicable, add the following below the CDDL
 * Header, with the fields enclosed by brackets [] replaced by your own identifying
 * information: "Portions Copyright [year] [name of copyright owner]".
 *
 * Copyright 2006-2010 Sun Microsystems, Inc.
 * Portions Copyright 2011-2016 ForgeRock AS.
 * Portions Copyright 2026 3A Systems, LLC
 */
package org.opends.server.replication.server;

import static org.opends.messages.ReplicationMessages.*;
import static org.opends.server.replication.protocol.ProtocolVersion.*;

import java.io.IOException;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

import org.forgerock.i18n.LocalizableMessage;
import org.forgerock.i18n.slf4j.LocalizedLogger;
import org.forgerock.opendj.ldap.DN;
import org.forgerock.opendj.ldap.ResultCode;
import org.opends.server.api.MonitorData;
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.protocol.ProtocolVersion;
import org.opends.server.replication.protocol.ReplServerStartMsg;
import org.opends.server.replication.protocol.ReplicationMsg;
import org.opends.server.replication.protocol.Session;
import org.opends.server.replication.protocol.StopMsg;
import org.opends.server.replication.protocol.TopologyMsg;
import org.opends.server.types.DirectoryException;
import org.opends.server.types.HostPort;

/**
 * This class defines a server handler, which handles all interaction with a
 * peer replication server.
 */
public class ReplicationServerHandler extends ServerHandler
{
  private static final LocalizedLogger logger = LocalizedLogger.getLoggerForThisClass();

  /** Properties filled only if remote server is a RS. */
  private String serverAddressURL;
  /**
   * The addresses the remote replication server is known by, built once, when its start
   * message names it: the connect thread compares them on every one of its passes, and
   * {@link HostPort} logs a name it cannot resolve each time it is built from one -- which
   * the fall back of {@code ReplicationServer.setServerURL()} to the host name of the
   * machine makes an ordinary thing for a peer to name. What the comparison of two of them
   * resolves is reported nowhere above trace, so building them once is what keeps that name
   * out of the error log rather than what keeps it out of the resolver.
   */
  private List<HostPort> addresses = Collections.emptyList();
  /**
   * This collection will contain as many elements as there are
   * LDAP servers connected to the remote replication server.
   */
  private final Map<Integer, LightweightServerHandler> remoteDirectoryServers = new ConcurrentHashMap<>();

  /**
   * Starts this handler based on a start message received from remote server.
   * @param inReplServerStartMsg The start msg provided by the remote server.
   * @return Whether the remote server requires encryption or not.
   * @throws DirectoryException When a problem occurs.
   */
  private boolean processStartFromRemote(
        ReplServerStartMsg inReplServerStartMsg)
        throws DirectoryException
  {
    try
    {
      short protocolVersion = getCompatibleVersion(inReplServerStartMsg
          .getVersion());
      session.setProtocolVersion(protocolVersion);
      generationId = inReplServerStartMsg.getGenerationId();
      serverId = inReplServerStartMsg.getServerId();
      serverURL = inReplServerStartMsg.getServerURL();
      setServerAddresses(serverURL);
      setBaseDNAndDomain(inReplServerStartMsg.getBaseDN(), false);
      setInitialServerState(inReplServerStartMsg.getServerState());
      setSendWindowSize(inReplServerStartMsg.getWindowSize());
      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();
      }
    }
    catch(Exception e)
    {
      LocalizableMessage message = LocalizableMessage.raw(e.getLocalizedMessage());
      throw new DirectoryException(ResultCode.OTHER, message);
    }
    return inReplServerStartMsg.getSSLEncryption();
  }

  /**
   * Takes the addresses the remote replication server is known by from the URL its start
   * message named: that URL, which is the address it is configured under, and the address
   * of the connection this session is held on, which is the interface that connection
   * happened to use rather than an identity.
   */
  private void setServerAddresses(String serverURL)
  {
    final HostPort namedAddress = HostPort.valueOf(serverURL);
    // Ensure correct formatting of IPv6 addresses by using a HostPort instance.
    final HostPort connectedAddress =
        new HostPort(session.getRemoteAddress().getHost(), namedAddress.getPort());
    serverAddressURL = connectedAddress.toString();
    addresses = Arrays.asList(namedAddress, connectedAddress);
  }

  /**
   * Sends a start message to the remote RS.
   *
   * @return The ReplServerStartMsg sent.
   * @throws IOException
   *           When an exception occurs.
   */
  private ReplServerStartMsg sendStartToRemote() throws IOException
  {
    ReplServerStartMsg outReplServerStartMsg = createReplServerStartMsg();
    send(outReplServerStartMsg);
    return outReplServerStartMsg;
  }

  /**
   * Creates a new handler object to a remote replication server.
   * @param session The session with the remote RS.
   * @param queueSize The queue size to manage updates to that RS.
   * @param replicationServer The hosting local RS object.
   * @param rcvWindowSize The receiving window size.
   */
  public ReplicationServerHandler(
      Session session,
      int queueSize,
      ReplicationServer replicationServer,
      int rcvWindowSize)
  {
    super(session, queueSize, replicationServer, rcvWindowSize);
  }

  /**
   * Connect the hosting RS to the RS represented by THIS handler
   * on an outgoing connection.
   * <p>
   * A handshake which does not complete is aborted rather than thrown out of here, and
   * three of its seven aborts carry no message at all: a peer which stops the handshake
   * answers with a {@link StopMsg}, which {@code Session.close()} publishes for every
   * abort of its own, and {@link #abortStart} logs nothing when the reason is null. The
   * caller is what is left to tell a connection from an attempt which only reached the
   * replication port of a peer.
   *
   * @param baseDN The baseDN
   * @param sslEncryption The sslEncryption requested to the remote RS.
   * @return {@code true} when the handshake completed and this handler is started,
   *         {@code false} when it was aborted.
   * @throws DirectoryException when an error occurs.
   */
  public boolean connect(DN baseDN, boolean sslEncryption)
      throws DirectoryException
  {
    // we are the initiator and decides of the encryption
    this.sslEncryption = sslEncryption;

    setBaseDNAndDomain(baseDN, false);

    try
    {
      lockDomainNoTimeout();

      // Read under the domain lock so the start message advertises any change
      // made by a previous handshake (handshakes serialize on this lock). The
      // field itself is only guarded by generationIDLock and lock-free
      // adopters can still move it — the arming CAS in
      // setDomainGenerationIdOnStart handles that residual race.
      localGenerationId = replicationServerDomain.getGenerationId();

      ReplServerStartMsg outReplServerStartMsg = sendStartToRemote();

      // Wait answer
      ReplicationMsg msg = session.receive();

      // Reject bad responses
      if (!(msg instanceof ReplServerStartMsg))
      {
        if (msg instanceof StopMsg)
        {
          // Remote replication server is probably shutting down or simultaneous
          // cross-connect detected.
          abortStart(null);
        }
        else
        {
          LocalizableMessage message = ERR_REPLICATION_PROTOCOL_MESSAGE_TYPE.get(msg
              .getClass().getCanonicalName(), "ReplServerStartMsg");
          abortStart(message);
        }
        return false;
      }

      processStartFromRemote((ReplServerStartMsg) msg);

      if (replicationServerDomain.isAlreadyConnectedToRS(this))
      {
        // Simultaneous cross connect.
        abortStart(null);
        return false;
      }

      /*
      Since we are going to send the topology message before having received
      one, we need to set the generation ID as soon as possible if it is
      currently uninitialized. See OpenDJ-121.
      */
      if (localGenerationId < 0 && generationId > 0)
      {
        setDomainGenerationIdOnStart(generationId);
      }

      logStartHandshakeSNDandRCV(outReplServerStartMsg,(ReplServerStartMsg)msg);

      // Until here session is encrypted then it depends on the negotiation
      // The session initiator decides whether to use SSL.
      if (!this.sslEncryption)
      {
        session.stopEncryption();
      }

      if (getProtocolVersion() > ProtocolVersion.REPLICATION_PROTOCOL_V1)
      {
        /*
        Only protocol version above V1 has a phase 2 handshake
        NOW PROCEED WITH SECOND PHASE OF HANDSHAKE:
        TopologyMsg then TopologyMsg (with a RS)

        Send our own TopologyMsg to remote RS
        */
        TopologyMsg outTopoMsg =
            replicationServerDomain.createTopologyMsgForRS();
        sendTopoInfo(outTopoMsg);

        // wait and process Topo from remote RS
        TopologyMsg inTopoMsg = waitAndProcessTopoFromRemoteRS();
        if (inTopoMsg == null)
        {
          // Simultaneous cross connect.
          abortStart(null);
          return false;
        }

        logTopoHandshakeSNDandRCV(outTopoMsg, inTopoMsg);

        /*
        FIXME: i think this should be done for all protocol version !!
        not only those > V1
        */
        replicationServerDomain.register(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);
      }

      logger.debug(INFO_REPLICATION_SERVER_CONNECTION_TO_RS, getReplicationServerId(), getServerId(),
          replicationServerDomain.getBaseDN(), session.getReadableRemoteAddress());

      super.finalizeStart();
      return true;
    }
    catch (IOException e)
    {
      logger.traceException(e);
      LocalizableMessage errMessage = ERR_RS_DISCONNECTED_DURING_HANDSHAKE.get(
          getReplicationServerId(),
          session.getReadableRemoteAddress());
      abortStart(errMessage);
      return false;
    }
    catch (DirectoryException e)
    {
      logger.traceException(e);
      abortStart(e.getMessageObject());
      return false;
    }
    catch (Exception e)
    {
      logger.traceException(e);
      abortStart(LocalizableMessage.raw(e.getLocalizedMessage()));
      return false;
    }
    finally
    {
      releaseDomainLock();
    }
  }

  /**
   * Starts the handler from a remote ReplServerStart message received from
   * the remote replication server.
   * @param inReplServerStartMsg The provided ReplServerStart message received.
   */
  public void startFromRemoteRS(ReplServerStartMsg inReplServerStartMsg)
  {
    localGenerationId = -1;
    try
    {
      // The initiator decides if the session is encrypted
      sslEncryption = processStartFromRemote(inReplServerStartMsg);

      lockDomainWithTimeout();

      if (replicationServerDomain.isAlreadyConnectedToRS(this))
      {
        abortStart(null);
        return;
      }

      this.localGenerationId = replicationServerDomain.getGenerationId();
      ReplServerStartMsg outReplServerStartMsg = sendStartToRemote();

      logStartHandshakeRCVandSND(inReplServerStartMsg, outReplServerStartMsg);

      /*
      until here session is encrypted then it depends on the negotiation
      The session initiator decides whether to use SSL.
      */
      if (!sslEncryption)
      {
        session.stopEncryption();
      }

      TopologyMsg inTopoMsg = null;
      if (getProtocolVersion() > ProtocolVersion.REPLICATION_PROTOCOL_V1)
      {
        /*
        Only protocol version above V1 has a phase 2 handshake
        NOW PROCEED WITH SECOND PHASE OF HANDSHAKE:
        TopologyMsg then TopologyMsg (with a RS)
        wait and process Topo from remote RS
        */
        inTopoMsg = waitAndProcessTopoFromRemoteRS();
        if (inTopoMsg == null)
        {
          // Simultaneous cross connect.
          abortStart(null);
          return;
        }

        // send our own TopologyMsg to remote RS
        TopologyMsg outTopoMsg = replicationServerDomain
            .createTopologyMsgForRS();
        sendTopoInfo(outTopoMsg);

        logTopoHandshakeRCVandSND(inTopoMsg, outTopoMsg);
      }
      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 (logger.isTraceEnabled())
          {
            logger.trace("In " + replicationServer.getMonitorInstanceName()
                + " " + this + " RS V1 with serverID=" + serverId
                + " is connected with the right generation ID");
          }
        } else
        {
          checkGenerationId();
        }
        /*
        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.
        */
      }

      replicationServerDomain.register(this);

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

      logger.debug(INFO_REPLICATION_SERVER_CONNECTION_FROM_RS, getReplicationServerId(), getServerId(),
          replicationServerDomain.getBaseDN(), session.getReadableRemoteAddress());

      super.finalizeStart();
    }
    catch (IOException e)
    {
      logger.traceException(e);
      abortStart(ERR_RS_DISCONNECTED_DURING_HANDSHAKE.get(
          inReplServerStartMsg.getServerId(), replicationServer.getServerId()));
    }
    catch (DirectoryException e)
    {
      logger.traceException(e);
      abortStart(e.getMessageObject());
    }
    catch (Exception e)
    {
      logger.traceException(e);
      abortStart(LocalizableMessage.raw(e.getLocalizedMessage()));
    }
    finally
    {
      releaseDomainLock();
    }
  }

  /**
   * Wait receiving the TopologyMsg from the remote RS and process it.
   * @return the topologyMsg received or {@code null} if stop was received.
   * @throws DirectoryException
   */
  private TopologyMsg waitAndProcessTopoFromRemoteRS()
      throws DirectoryException
  {
    ReplicationMsg msg;
    try
    {
      msg = session.receive();
    }
    catch(Exception e)
    {
      LocalizableMessage message = LocalizableMessage.raw(e.getLocalizedMessage());
      throw new DirectoryException(ResultCode.OTHER, message);
    }

    if (!(msg instanceof TopologyMsg))
    {
      if (msg instanceof StopMsg)
      {
        // Remote replication server is probably shutting down, or cross
        // connection attempt.
        return null;
      }

      LocalizableMessage message = ERR_REPLICATION_PROTOCOL_MESSAGE_TYPE.get(
          msg.getClass().getCanonicalName(), "TopologyMsg");
      throw new DirectoryException(ResultCode.OTHER, message);
    }

    // Remote RS sent his topo msg
    TopologyMsg inTopoMsg = (TopologyMsg) msg;

    /* Store remote RS weight if it has one.
     * For protocol version < 4, use default value of 1 for weight
     */
    if (getProtocolVersion() >= ProtocolVersion.REPLICATION_PROTOCOL_V4)
    {
      // List should only contain RS info for sender
      RSInfo rsInfo = inTopoMsg.getRsInfos().get(0);
      weight = rsInfo.getWeight();
    }

    /*
    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 (logger.isTraceEnabled())
      {
        logger.trace("In " + replicationServer.getMonitorInstanceName()
            + " RS with serverID=" + serverId
            + " is connected with the right generation ID, same as local ="
            + generationId);
      }
    }
    else
    {
      checkGenerationId();
    }

    return inTopoMsg;
  }

  /**
   * Checks local generation ID against the remote RS one,
   * and logs Warning messages if needed.
   */
  private void checkGenerationId()
  {
    if (localGenerationId <= 0)
    {
      // 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
      setDomainGenerationIdOnStart(generationId);
      return;
    }

    // the local RS is initialized
    if (generationId > 0
        // the remote RS is initialized. If not, there's nothing to do anyway.
        && generationId != localGenerationId)
    {
      /* Either:
       *
       * 1) The 2 RS have different generationID
       * replicationServerDomain.getGenerationIdSavedStatus() == true
       *
       * if the present RS has received changes regarding its gen ID and so will
       * not change without a reset then we are just degrading the peer.
       *
       * 2) This RS has never received any changes for the current 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 null state ?
       * replicationServerDomain.setGenerationId(generationId, false);
       */
      logger.warn(WARN_BAD_GENERATION_ID_FROM_RS, serverId, session.getReadableRemoteAddress(), generationId,
          getBaseDN(), getReplicationServerId(), localGenerationId);
    }
  }

  /** {@inheritDoc} */
  @Override
  public boolean isDataServer()
  {
    return false;
  }

  /**
   * 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 remoteDirectoryServers
   *                connected to this ServerHandler.
   */
  public void addDSInfos(List<DSInfo> dsInfos)
  {
    synchronized (remoteDirectoryServers)
    {
      for (LightweightServerHandler ls : remoteDirectoryServers.values())
      {
        dsInfos.add(ls.toDSInfo());
      }
    }
  }

  /**
   * Shutdown This ServerHandler.
   */
  @Override
  public void shutdown()
  {
    super.shutdown();
    clearRemoteLSHandlers();
  }

  private void clearRemoteLSHandlers()
  {
    synchronized (remoteDirectoryServers)
    {
      for (LightweightServerHandler lsh : remoteDirectoryServers.values())
      {
        lsh.stopHandler();
      }
      remoteDirectoryServers.clear();
    }
  }

  /**
   * Stores topology information received from a peer RS and that must be kept
   * in RS handler.
   *
   * @param topoMsg The received topology message
   */
  public void processTopoInfoFromRS(TopologyMsg topoMsg)
  {
    // List should only contain RS info for sender
    final RSInfo rsInfo = topoMsg.getRsInfos().get(0);
    generationId = rsInfo.getGenerationId();
    groupId = rsInfo.getGroupId();
    weight = rsInfo.getWeight();

    synchronized (remoteDirectoryServers)
    {
      clearRemoteLSHandlers();

      // Creates the new structure according to the message received.
      for (DSInfo dsInfo : topoMsg.getReplicaInfos().values())
      {
        // For each DS connected to the peer RS
        DSInfo clonedDSInfo = dsInfo.cloneWithReplicationServerId(serverId);
        LightweightServerHandler lsh =
            new LightweightServerHandler(this, clonedDSInfo);
        lsh.startHandler();
        remoteDirectoryServers.put(lsh.getServerId(), lsh);
      }
    }
  }

  /**
   * When this handler is connected to a replication server, specifies if
   * a wanted server is connected to this replication server.
   *
   * @param serverId 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(int serverId)
  {
    synchronized (remoteDirectoryServers)
    {
      for (LightweightServerHandler server : remoteDirectoryServers.values())
      {
        if (serverId == 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()
  {
    return !remoteDirectoryServers.isEmpty();
  }

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

  /** {@inheritDoc} */
  @Override
  public String getMonitorInstanceName()
  {
    return "Connected replication server RS(" + serverId + ") " + serverURL
        + ",cn=" + replicationServerDomain.getMonitorInstanceName();
  }

  @Override
  public MonitorData getMonitorData()
  {
    MonitorData attributes = super.getMonitorData();

    ReplicationDomainMonitorData md = replicationServerDomain.getDomainMonitorData();
    attributes.add("Replication-Server", serverURL);
    attributes.add("missing-changes", md.getMissingChangesRS(serverId));

    ServerState state = md.getRSStates(serverId);
    if (state != null)
    {
      attributes.add("server-state", state.toStringSet());
    }

    return attributes;
  }

  /** {@inheritDoc} */
  @Override
  public String toString()
  {
    if (serverId != 0)
    {
      return "Replication server RS(" + serverId + ") for domain \""
          + replicationServerDomain.getBaseDN() + "\"";
    }
    return "Unknown server";
  }

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

  /**
   * 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;
  }

  /**
   * Returns whether the remote replication server of this handler is the one the provided
   * handler holds a session with.
   * <p>
   * Either of the two addresses a remote server is known by identifies it, and the one
   * which does depends on where its connection came from: a server reachable at more than
   * one address -- a multi homed host, or a NAT where the address a peer connects
   * <i>from</i> is not the address it is configured <i>as</i> -- is registered under the
   * address of whichever interface the session used, so two sessions with one such server
   * carry two different addresses. What both of them do carry is the address that server
   * names in its start messages, which is the address it is configured under.
   *
   * @param other
   *          the handler to compare the remote server of this one with
   * @return {@code true} if both handlers hold a session with the same replication server
   */
  boolean isSameServerAs(ReplicationServerHandler other)
  {
    for (HostPort address : other.addresses)
    {
      if (isServerAt(address))
      {
        return true;
      }
    }
    return false;
  }

  /**
   * Returns whether the remote replication server of this handler is the one configured at
   * the provided address.
   * <p>
   * Both addresses it is known by are compared, because either of them may be the
   * configured one: the address the remote server names is the address it is configured
   * under in its own configuration, which is the one the rest of the topology configures it
   * at as well, while the address its session came from is the only one known of a server
   * which names an address this configuration does not use.
   * <p>
   * A name neither end can resolve is compared as the name it is, which is what
   * {@link HostPort#equals(Object)} does with it: {@link HostPort#isEquivalentTo(HostPort)}
   * resolves both hosts and answers {@code false} for a name it cannot resolve, even
   * against that same name. The peer which names one is the peer of the fall back of
   * {@code ReplicationServer.setServerURL()}, whose own host name the rest of the topology
   * has no reason to resolve, and it is the peer the addresses are there for. What that
   * gives up is two servers which share a server id and both name one unresolvable name on
   * one port: they are read as one, as two servers which name the same resolvable address
   * are under the other arm.
   * <p>
   * A pair the names alone do not answer is resolved on every call, and the handshake road
   * makes those calls under the domain lock: both {@code startFromRemoteRS()} and the
   * connect road hold that lock across
   * {@link ReplicationServerDomain#isAlreadyConnectedToRS(ReplicationServerHandler)}, so a
   * resolver which does not answer holds the domain for its own timeout on the road where
   * two handlers of one server id name different hosts. Only replication servers reach it:
   * the handshake of a data server makes no such comparison.
   *
   * @param address
   *          a configured address of a replication server
   * @return {@code true} if the remote server of this handler answers to that address
   */
  boolean isServerAt(HostPort address)
  {
    for (HostPort known : addresses)
    {
      if (address.equals(known) || address.isEquivalentTo(known))
      {
        return true;
      }
    }
    return false;
  }

  /**
   * Receives a topology msg.
   * @param topoMsg The message received.
   * @throws DirectoryException when it occurs.
   * @throws IOException when it occurs.
   */
  public void receiveTopoInfoFromRS(TopologyMsg topoMsg)
  throws DirectoryException, IOException
  {
    replicationServerDomain.receiveTopoInfoFromRS(topoMsg, this, true);
  }

}