/*
 * 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 legal-notices/CDDLv1_0.txt
 * or http://forgerock.org/license/CDDLv1.0.html.
 * 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 legal-notices/CDDLv1_0.txt.
 * 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.
 *      Portions Copyright 2011-2014 ForgeRock AS
 */
package org.opends.server.replication.server;

import java.net.SocketException;

import org.forgerock.i18n.LocalizableMessage;
import org.opends.server.api.DirectoryThread;
import org.forgerock.i18n.slf4j.LocalizedLogger;
import org.opends.server.replication.common.ServerStatus;
import org.opends.server.replication.protocol.Session;
import org.opends.server.replication.protocol.UpdateMsg;

import static org.opends.messages.ReplicationMessages.*;
import static org.opends.server.loggers.ErrorLogger.*;
import static org.opends.server.util.StaticUtils.*;

/**
 * This class defines a server writer, which is used to send changes to a
 * directory server.
 */
public class ServerWriter extends DirectoryThread
{
  private static final LocalizedLogger logger = LocalizedLogger.getLoggerForThisClass();

  private final Session session;
  private final ServerHandler handler;
  private final ReplicationServerDomain replicationServerDomain;



  /**
   * Create a ServerWriter. Then ServerWriter then waits on the ServerHandler
   * for new updates and forward them to the server
   *
   * @param session
   *          the Session that will be used to send updates.
   * @param handler
   *          handler for which the ServerWriter is created.
   * @param replicationServerDomain
   *          The ReplicationServerDomain of this ServerWriter.
   */
  public ServerWriter(Session session, ServerHandler handler,
      ReplicationServerDomain replicationServerDomain)
  {
    // Session may be null for ECLServerWriter.
    super("Replication server RS(" + handler.getReplicationServerId()
        + ") writing to " + handler + " at "
        + (session != null ? session.getReadableRemoteAddress() : "unknown"));

    this.session = session;
    this.handler = handler;
    this.replicationServerDomain = replicationServerDomain;
  }

  /**
   * Run method for the ServerWriter.
   * Loops waiting for changes from the ReplicationServerDomain and forward them
   * to the other servers
   */
  @Override
  public void run()
  {
    if (logger.isTraceEnabled())
    {
      logger.trace(getName() + " starting");
    }

    LocalizableMessage errMessage = null;
    try
    {
      while (true)
      {
        UpdateMsg update = replicationServerDomain.take(this.handler);
        if (update == null)
        {
          // this connection is closing
          errMessage = LocalizableMessage.raw(
           "Connection closure: null update returned by domain.");
          return;
        }

        // Ignore updates in some cases
        long referenceGenerationId = replicationServerDomain.getGenerationId();
        if (handler.isDataServer())
        {
          /**
           * Ignore updates to DS in bad BAD_GENID_STATUS or FULL_UPDATE_STATUS
           *
           * The RSD lock should not be taken here as it is acceptable to have a
           * delay between the time the server has a wrong status and the fact
           * we detect it: the updates that succeed to pass during this time
           * will have no impact on remote server. But it is interesting to not
           * saturate uselessly the network if the updates are not necessary so
           * this check to stop sending updates is interesting anyway. Not
           * taking the RSD lock allows to have better performances in normal
           * mode (most of the time).
           */
          ServerStatus dsStatus = handler.getStatus();
          if (dsStatus == ServerStatus.BAD_GEN_ID_STATUS
              || dsStatus == ServerStatus.FULL_UPDATE_STATUS)
          {
            if (dsStatus == ServerStatus.BAD_GEN_ID_STATUS)
            {
              logError(WARN_IGNORING_UPDATE_TO_DS_BADGENID.get(
                  handler.getReplicationServerId(),
                  update.getCSN().toString(),
                  handler.getBaseDNString(), handler.getServerId(),
                  session.getReadableRemoteAddress(),
                  handler.getGenerationId(),
                  referenceGenerationId));
            }
            else if (dsStatus == ServerStatus.FULL_UPDATE_STATUS)
            {
              logError(WARN_IGNORING_UPDATE_TO_DS_FULLUP.get(
                  handler.getReplicationServerId(),
                  update.getCSN().toString(),
                  handler.getBaseDNString(), handler.getServerId(),
                  session.getReadableRemoteAddress()));
            }
            continue;
          }
        }
        else
        {
          /**
           * Ignore updates to RS with bad gen id
           * (no system managed status for a RS)
           */
          if (referenceGenerationId != handler.getGenerationId()
              || referenceGenerationId == -1
              || handler.getGenerationId() == -1)
          {
            logError(
                WARN_IGNORING_UPDATE_TO_RS.get(
                    handler.getReplicationServerId(),
                    update.getCSN().toString(),
                    handler.getBaseDNString(),
                    handler.getServerId(),
                    session.getReadableRemoteAddress(),
                    handler.getGenerationId(),
                    referenceGenerationId));
            continue;
          }
        }

        // Publish the update to the remote server using a protocol version he
        // it supports
        session.publish(update);
      }
    }
    catch (SocketException e)
    {
      /*
       * The remote host has disconnected and this particular Tree is going to
       * be removed, just ignore the exception and let the thread die as well
       */
      errMessage = handler.getBadlyDisconnectedErrorMessage();
      logError(errMessage);
    }
    catch (Exception e)
    {
      /*
       * An unexpected error happened.
       * Log an error and close the connection.
       */
      errMessage = ERR_WRITER_UNEXPECTED_EXCEPTION.get(handler +
                        " " +  stackTraceToSingleLineString(e));
      logError(errMessage);
    }
    finally {
      session.close();
      replicationServerDomain.stopServer(handler, false);
      if (logger.isTraceEnabled())
      {
        logger.trace(getName() + " stopped " + errMessage);
      }
    }
  }
}