/*
 * 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 2009 Sun Microsystems, Inc.
 *      Portions copyright 2011-2014 ForgeRock AS
 */
package org.opends.server.replication.server;

import java.io.IOException;
import java.net.SocketException;

import org.opends.server.core.DirectoryServer;
import org.opends.server.core.PersistentSearch;
import org.opends.server.loggers.debug.DebugTracer;
import org.opends.server.replication.protocol.DoneMsg;
import org.opends.server.replication.protocol.ECLUpdateMsg;
import org.opends.server.replication.protocol.Session;
import org.opends.server.replication.service.DSRSShutdownSync;
import org.opends.server.types.DebugLogLevel;
import org.opends.server.types.DirectoryException;
import org.opends.server.types.Entry;
import org.opends.server.workflowelement.externalchangelog.ECLSearchOperation;
import org.opends.server.workflowelement.externalchangelog.ECLWorkflowElement;

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

/**
 * This class defines a server writer, which is used to send changes to a
 * directory server.
 */
class ECLServerWriter extends ServerWriter
{
  /**
   * The tracer object for the debug logger.
   */
  private static final DebugTracer TRACER = getTracer();

  private final Session session;
  private final ECLServerHandler handler;
  private final ReplicationServerDomain replicationServerDomain;
  private boolean suspended;
  private volatile boolean shutdown;
  private final PersistentSearch mypsearch;

  /**
   * Create a ServerWriter.
   *
   * @param session     the Session that will be used to send updates.
   * @param handler     ECL handler for which the ServerWriter is created.
   * @param replicationServerDomain the ReplicationServerDomain of this
   *                    ServerWriter.
   */
  ECLServerWriter(Session session, ECLServerHandler handler,
      ReplicationServerDomain replicationServerDomain)
  {
    super(session, handler, replicationServerDomain, new DSRSShutdownSync());

    setName("Replication ECL Writer Thread for operation " +
        handler.getOperationId());

    this.session = session;
    this.handler = handler;
    this.replicationServerDomain = replicationServerDomain;
    this.suspended = false;
    this.shutdown = false;
    this.mypsearch = findPersistentSearch(handler);
  }

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

  /**
   * The writer will start suspended by the Handler for the CL
   * waiting for the startCLSessionMsg. Then it may be
   * suspended between 2 jobs, each job being a separate search.
   */
  private synchronized void suspendWriter()
  {
    suspended = true;
  }

  /**
   * Resume the writer.
   */
  synchronized void resumeWriter()
  {
    suspended = false;
    notify();
  }

  /**
   * Run method for the ServerWriter.
   * Loops waiting for changes from the ReplicationServerDomain and
   * forward them to the other servers
   */
  @Override
  public void run()
  {
    try
    {
      while (true)
      {
        // wait to be resumed or shutdown
        if (suspended && !shutdown)
        {
          synchronized(this)
          {
            wait();
          }
        }

        if (shutdown)
          return;

        // Not suspended
        doIt();

        if (shutdown)
          return;

        suspendWriter();
      }
    }
    catch (SocketException e)
    {
      // Just ignore the exception and let the thread die as well.
      // session is always null if a socket exception has occurred.
      if (session != null)
      {
        logError(handler.getBadlyDisconnectedErrorMessage());
      }
    }
    catch (Exception e)
    {
      // An unexpected error happened.
      // Log an error and close the connection.
      logError(ERR_WRITER_UNEXPECTED_EXCEPTION.get(
          handler + " " + stackTraceToSingleLineString(e)));
    }
    finally
    {
      if (session!=null)
      {
        session.close();
      }
      if (replicationServerDomain != null)
        replicationServerDomain.stopServer(handler, false);
    }
  }

  /**
   * Loop getting changes from the domain and publishing them either to
   * the provided session or to the ECL session interface.
   * @throws IOException when raised (connection closure)
   * @throws InterruptedException when raised
   */
  private void doIt() throws IOException, InterruptedException
  {
    while (true)
    {
      if (shutdown || suspended)
      {
        return;
      }

      ECLUpdateMsg update = null;
      try
      {
        update = handler.takeECLUpdate();
      }
      catch(DirectoryException de)
      {
        TRACER.debugCaught(DebugLogLevel.ERROR, de);
      }

      if (update == null)
      {
        if (session != null && handler.isInitPhaseDone())
        {
          // session is null in pusherOnly mode
          // Done is used to end phase 1
          session.publish(new DoneMsg(
              handler.getReplicationServerId(), handler.getServerId()));
        }

        if (handler.isNonPersistent())
        { // publishing is normally stopped here...
          break;
        }

        // ...except if we are in persistent search
        Thread.sleep(200);
      }
      else
      {
        // Publish the update to the remote server using a protocol version it
        // supports
        publish(update);
        update = null;
      }
    }
  }

  /**
   * Shutdown the writer.
   */
  synchronized void shutdownWriter()
  {
    shutdown = true;
    notify();
  }

  /**
   * Publish a change either on the protocol session or to a persistent search.
   */
  private void publish(ECLUpdateMsg msg) throws IOException
  {
    if (debugEnabled())
      TRACER.debugInfo(getName() + " publishes msg=[" + msg + "]");

    if (session != null)
    {
      session.publish(msg);
    }
    else if (mypsearch != null)
    {
      try
      {
        Entry eclEntry = ECLSearchOperation.createEntryFromMsg(msg);
        mypsearch.processAdd(eclEntry, -1);
      }
      catch (Exception e)
      {
        logError(ERR_WRITER_UNEXPECTED_EXCEPTION.get(
            handler + " " + stackTraceToSingleLineString(e)));
        mypsearch.cancel();
      }
    }
  }
}