/*
 * 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/opendj3/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
 * trunk/opendj3/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 2011-2012 ForgeRock AS
 */

package org.forgerock.opendj.ldap;



import static org.forgerock.opendj.ldap.CoreMessages.*;
import static org.forgerock.opendj.ldap.ErrorResultException.newErrorResult;

import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;

import org.forgerock.i18n.LocalizableMessage;
import org.forgerock.opendj.ldap.requests.*;
import org.forgerock.opendj.ldap.responses.*;

import com.forgerock.opendj.util.Validator;



/**
 * An adapter which converts a {@code RequestHandlerFactory} into a
 * {@code ServerConnectionFactory}.
 *
 * @param <C>
 *          The type of client context.
 */
final class RequestHandlerFactoryAdapter<C> implements
    ServerConnectionFactory<C, Integer>
{
  /**
   * Request context implementation.
   */
  private static class RequestContextImpl<S extends Result, H extends ResultHandler<? super S>>
      implements RequestContext, ResultHandler<S>
  {

    // Adapter class which invokes cancel result handlers with correct result
    // type.
    private static final class ExtendedResultHandlerHolder<R extends ExtendedResult>
    {
      private final ExtendedRequest<R> request;
      private final ResultHandler<? super R> resultHandler;



      private ExtendedResultHandlerHolder(final ExtendedRequest<R> request,
          final ResultHandler<? super R> resultHandler)
      {
        this.request = request;
        this.resultHandler = resultHandler;
      }



      private void handleSuccess()
      {
        final R cancelResult = request.getResultDecoder()
            .newExtendedErrorResult(ResultCode.SUCCESS, "", "");
        resultHandler.handleResult(cancelResult);
      }



      private void handleTooLate()
      {
        final R cancelResult = request.getResultDecoder()
            .newExtendedErrorResult(ResultCode.TOO_LATE, "", "");
        resultHandler.handleErrorResult(ErrorResultException
            .newErrorResult(cancelResult));
      }
    }



    private static enum RequestState
    {
      // Request active
      PENDING,

      // Request active, cancel requested
      CANCEL_REQUESTED,

      // Request active, too late to cancel
      TOO_LATE,

      // Result sent, not cancelled
      RESULT_SENT,

      // Result sent, was cancelled
      CANCELLED;
    }



    private final int messageID;

    // Cancellation state guarded by lock.
    private final Object stateLock = new Object();

    // These should be notified when a cancel request arrives, at most once.
    private List<CancelRequestListener> cancelRequestListeners = null;

    // These should be notified when the result is set.
    private List<ExtendedResultHandlerHolder<?>> cancelResultHandlers = null;

    private RequestState state = RequestState.PENDING;

    private LocalizableMessage cancelRequestReason = null;

    private boolean sendResult = true;

    private final boolean isCancelSupported;

    private final ServerConnectionImpl<?> clientConnection;

    protected final H resultHandler;



    protected RequestContextImpl(
        final ServerConnectionImpl<?> clientConnection, final H resultHandler,
        final int messageID, final boolean isCancelSupported)
    {
      this.clientConnection = clientConnection;
      this.resultHandler = resultHandler;
      this.messageID = messageID;
      this.isCancelSupported = isCancelSupported;
    }



    /**
     * {@inheritDoc}
     */
    @Override
    public void addCancelRequestListener(final CancelRequestListener listener)
    {
      Validator.ensureNotNull(listener);

      boolean invokeImmediately = false;
      synchronized (stateLock)
      {
        switch (state)
        {
        case PENDING:
          if (cancelRequestListeners == null)
          {
            cancelRequestListeners = new LinkedList<CancelRequestListener>();
          }
          cancelRequestListeners.add(listener);
          break;
        case CANCEL_REQUESTED:
          // Signal immediately outside lock.
          invokeImmediately = true;
          break;
        case TOO_LATE:
        case RESULT_SENT:
        case CANCELLED:
          // No point in registering the callback since the request can never be
          // cancelled now.
          break;
        }
      }

      if (invokeImmediately)
      {
        listener.handleCancelRequest(cancelRequestReason);
      }
    }



    /**
     * {@inheritDoc}
     */
    @Override
    public void checkIfCancelled(final boolean signalTooLate)
        throws CancelledResultException
    {
      synchronized (stateLock)
      {
        switch (state)
        {
        case PENDING:
          // No cancel request, so no handlers, just switch state.
          if (signalTooLate)
          {
            cancelRequestListeners = null;
            state = RequestState.TOO_LATE;
          }
          break;
        case CANCEL_REQUESTED:
          // Don't change state: let the handler ack the cancellation request.
          throw (CancelledResultException) newErrorResult(ResultCode.CANCELLED,
              cancelRequestReason.toString());
        case TOO_LATE:
          // Already too late. Nothing to do.
          break;
        case RESULT_SENT:
        case CANCELLED:
          // This should not happen - could throw an illegal state exception?
          break;
        }
      }
    }



    /**
     * {@inheritDoc}
     */
    @Override
    public int getMessageID()
    {
      return messageID;
    }



    /**
     * {@inheritDoc}
     */
    @Override
    public void handleErrorResult(final ErrorResultException error)
    {
      if (clientConnection.removePendingRequest(this))
      {
        if (setResult(error.getResult()))
        {
          // FIXME: we must invoke the result handler even when abandoned so
          // that chained result handlers may clean up, log, etc. We really need
          // to signal that the result must not be sent to the client.
        }
        resultHandler.handleErrorResult(error);
      }
    }



    /**
     * {@inheritDoc}
     */
    @Override
    public void handleResult(final S result)
    {
      if (clientConnection.removePendingRequest(this))
      {
        if (setResult(result))
        {
          // FIXME: we must invoke the result handler even when abandoned so
          // that chained result handlers may clean up, log, etc. We really need
          // to signal that the result must not be sent to the client.
        }
        resultHandler.handleResult(result);
      }
    }



    /**
     * {@inheritDoc}
     */
    @Override
    public void removeCancelRequestListener(final CancelRequestListener listener)
    {
      Validator.ensureNotNull(listener);

      synchronized (stateLock)
      {
        if (cancelRequestListeners != null)
        {
          cancelRequestListeners.remove(listener);
        }
      }
    }



    private <R extends ExtendedResult> void cancel(
        final LocalizableMessage reason,
        final ExtendedRequest<R> cancelRequest,
        final ResultHandler<? super R> cancelResultHandler,
        final boolean sendResult)
    {
      Validator.ensureNotNull(reason);

      if (!isCancelSupported)
      {
        if (cancelResultHandler != null)
        {
          final Result result = Responses
              .newGenericExtendedResult(ResultCode.CANNOT_CANCEL);
          cancelResultHandler.handleErrorResult(newErrorResult(result));
        }
        return;
      }

      List<CancelRequestListener> tmpListeners = null;
      boolean invokeResultHandler = false;
      boolean resultHandlerIsSuccess = false;

      synchronized (stateLock)
      {
        switch (state)
        {
        case PENDING:
          // Switch to CANCEL_REQUESTED state.
          cancelRequestReason = reason;
          if (cancelResultHandler != null)
          {
            cancelResultHandlers = new LinkedList<ExtendedResultHandlerHolder<?>>();
            cancelResultHandlers.add(new ExtendedResultHandlerHolder<R>(
                cancelRequest, cancelResultHandler));
          }
          tmpListeners = cancelRequestListeners;
          cancelRequestListeners = null;
          state = RequestState.CANCEL_REQUESTED;
          this.sendResult &= sendResult;
          break;
        case CANCEL_REQUESTED:
          // Cancel already request so listeners already invoked.
          if (cancelResultHandler != null)
          {
            if (cancelResultHandlers == null)
            {
              cancelResultHandlers = new LinkedList<ExtendedResultHandlerHolder<?>>();
            }
            cancelResultHandlers.add(new ExtendedResultHandlerHolder<R>(
                cancelRequest, cancelResultHandler));
          }
          break;
        case TOO_LATE:
        case RESULT_SENT:
          // Cannot cancel, so invoke result handler immediately outside of
          // lock.
          if (cancelResultHandler != null)
          {
            invokeResultHandler = true;
            resultHandlerIsSuccess = false;
          }
          break;
        case CANCELLED:
          // Multiple cancellation attempts. Clients should not do this, but the
          // cancel will effectively succeed immediately, so invoke result
          // handler immediately outside of lock.
          if (cancelResultHandler != null)
          {
            invokeResultHandler = true;
            resultHandlerIsSuccess = true;
          }
          break;
        }
      }

      // Invoke listeners outside of lock.
      if (tmpListeners != null)
      {
        for (final CancelRequestListener listener : tmpListeners)
        {
          listener.handleCancelRequest(reason);
        }
      }

      if (invokeResultHandler)
      {
        if (resultHandlerIsSuccess)
        {
          final R result = cancelRequest.getResultDecoder()
              .newExtendedErrorResult(ResultCode.SUCCESS, "", "");
          cancelResultHandler.handleResult(result);
        }
        else
        {
          final Result result = Responses
              .newGenericExtendedResult(ResultCode.TOO_LATE);
          cancelResultHandler.handleErrorResult(ErrorResultException
              .newErrorResult(result));
        }
      }
    }



    /**
     * Sets the result associated with this request context and updates the
     * state accordingly.
     *
     * @param result
     *          The result.
     */
    private boolean setResult(final Result result)
    {
      List<ExtendedResultHandlerHolder<?>> tmpHandlers = null;
      boolean isCancelled = false;
      boolean maySendResult;

      synchronized (stateLock)
      {
        maySendResult = sendResult;

        switch (state)
        {
        case PENDING:
        case TOO_LATE:
          // Switch to appropriate final state.
          if (!result.getResultCode().equals(ResultCode.CANCELLED))
          {
            state = RequestState.RESULT_SENT;
          }
          else
          {
            state = RequestState.CANCELLED;
          }
          break;
        case CANCEL_REQUESTED:
          // Switch to appropriate final state and invoke any cancel request
          // handlers.
          if (!result.getResultCode().equals(ResultCode.CANCELLED))
          {
            state = RequestState.RESULT_SENT;
          }
          else
          {
            state = RequestState.CANCELLED;
          }

          isCancelled = (state == RequestState.CANCELLED);
          tmpHandlers = cancelResultHandlers;
          cancelResultHandlers = null;
          break;
        case RESULT_SENT:
        case CANCELLED:
          // This should not happen - could throw an illegal state exception?
          maySendResult = false; // Prevent sending multiple results.
          break;
        }
      }

      // Invoke handlers outside of lock.
      if (tmpHandlers != null)
      {
        for (final ExtendedResultHandlerHolder<?> handler : tmpHandlers)
        {
          if (isCancelled)
          {
            handler.handleSuccess();
          }
          else
          {
            handler.handleTooLate();
          }
        }
      }

      return maySendResult;
    }
  }



  /**
   * Search request context implementation.
   */
  private final static class SearchRequestContextImpl extends
      RequestContextImpl<Result, SearchResultHandler> implements
      SearchResultHandler
  {

    private SearchRequestContextImpl(
        final ServerConnectionImpl<?> clientConnection,
        final SearchResultHandler resultHandler, final int messageID,
        final boolean isCancelSupported)
    {
      super(clientConnection, resultHandler, messageID, isCancelSupported);
    }



    /**
     * {@inheritDoc}
     */
    @Override
    public boolean handleEntry(final SearchResultEntry entry)
    {
      return resultHandler.handleEntry(entry);
    }



    /**
     * {@inheritDoc}
     */
    @Override
    public boolean handleReference(final SearchResultReference reference)
    {
      return resultHandler.handleReference(reference);
    }
  }



  private static final class ServerConnectionImpl<C> implements
      ServerConnection<Integer>
  {
    private final RequestHandler<RequestContext> requestHandler;
    private final AtomicBoolean isClosed = new AtomicBoolean();
    private final ConcurrentHashMap<Integer, RequestContextImpl<?, ?>> pendingRequests =
        new ConcurrentHashMap<Integer, RequestContextImpl<?, ?>>();



    private ServerConnectionImpl(
        final RequestHandler<RequestContext> requestHandler)
    {
      this.requestHandler = requestHandler;
    }



    /**
     * {@inheritDoc}
     */
    @Override
    public void handleAbandon(final Integer messageID,
        final AbandonRequest request)
    {
      final RequestContextImpl<?, ?> abandonedRequest = getPendingRequest(request
          .getRequestID());
      if (abandonedRequest != null)
      {
        final LocalizableMessage abandonReason = INFO_CANCELED_BY_ABANDON_REQUEST
            .get(messageID);
        abandonedRequest.cancel(abandonReason, null, null, false);
      }
    }



    /**
     * {@inheritDoc}
     */
    @Override
    public void handleAdd(final Integer messageID, final AddRequest request,
        final IntermediateResponseHandler intermediateResponseHandler,
        final ResultHandler<? super Result> resultHandler)
    {
      final RequestContextImpl<Result, ResultHandler<? super Result>> requestContext =
          new RequestContextImpl<Result, ResultHandler<? super Result>>(
          this, resultHandler, messageID, true);
      if (addPendingRequest(requestContext))
      {
        requestHandler.handleAdd(requestContext, request,
            intermediateResponseHandler, requestContext);
      }
    }



    /**
     * {@inheritDoc}
     */
    @Override
    public void handleBind(final Integer messageID, final int version,
        final BindRequest request,
        final IntermediateResponseHandler intermediateResponseHandler,
        final ResultHandler<? super BindResult> resultHandler)
    {
      final RequestContextImpl<BindResult, ResultHandler<? super BindResult>> requestContext =
          new RequestContextImpl<BindResult, ResultHandler<? super BindResult>>(
          this, resultHandler, messageID, false);
      if (addPendingRequest(requestContext))
      {
        requestHandler.handleBind(requestContext, version, request,
            intermediateResponseHandler, requestContext);
      }
    }



    /**
     * {@inheritDoc}
     */
    @Override
    public void handleCompare(final Integer messageID,
        final CompareRequest request,
        final IntermediateResponseHandler intermediateResponseHandler,
        final ResultHandler<? super CompareResult> resultHandler)
    {
      final RequestContextImpl<CompareResult, ResultHandler<? super CompareResult>> requestContext =
          new RequestContextImpl<CompareResult, ResultHandler<? super CompareResult>>(
          this, resultHandler, messageID, true);
      if (addPendingRequest(requestContext))
      {
        requestHandler.handleCompare(requestContext, request,
            intermediateResponseHandler, requestContext);
      }
    }



    /**
     * {@inheritDoc}
     */
    @Override
    public void handleConnectionClosed(final Integer messageID,
        final UnbindRequest request)
    {
      final LocalizableMessage cancelReason = INFO_CANCELED_BY_CLIENT_DISCONNECT
          .get();
      doClose(cancelReason);
    }



    /**
     * {@inheritDoc}
     */
    @Override
    public void handleConnectionDisconnected(final ResultCode resultCode,
        final String message)
    {
      final LocalizableMessage cancelReason = INFO_CANCELED_BY_SERVER_DISCONNECT
          .get();
      doClose(cancelReason);
    }



    /**
     * {@inheritDoc}
     */
    @Override
    public void handleConnectionError(final Throwable error)
    {
      final LocalizableMessage cancelReason = INFO_CANCELED_BY_CLIENT_ERROR
          .get();
      doClose(cancelReason);
    }



    /**
     * {@inheritDoc}
     */
    @Override
    public void handleDelete(final Integer messageID,
        final DeleteRequest request,
        final IntermediateResponseHandler intermediateResponseHandler,
        final ResultHandler<? super Result> resultHandler)
    {
      final RequestContextImpl<Result, ResultHandler<? super Result>> requestContext =
          new RequestContextImpl<Result, ResultHandler<? super Result>>(
          this, resultHandler, messageID, true);
      if (addPendingRequest(requestContext))
      {
        requestHandler.handleDelete(requestContext, request,
            intermediateResponseHandler, requestContext);
      }
    }



    /**
     * {@inheritDoc}
     */
    @Override
    public <R extends ExtendedResult> void handleExtendedRequest(
        final Integer messageID, final ExtendedRequest<R> request,
        final IntermediateResponseHandler intermediateResponseHandler,
        final ResultHandler<? super R> resultHandler)
    {
      if (request.getOID().equals(CancelExtendedRequest.OID))
      {
        // Decode the request as a cancel request.
        CancelExtendedRequest cancelRequest;
        try
        {
          cancelRequest = CancelExtendedRequest.DECODER.decodeExtendedRequest(
              request, new DecodeOptions());
        }
        catch (final DecodeException e)
        {
          // Couldn't decode a cancel request.
          resultHandler.handleErrorResult(newErrorResult(
              ResultCode.PROTOCOL_ERROR, e.getLocalizedMessage()));
          return;
        }

        // Register the request in the pending requests table. Even though
        // this request cannot be cancelled, it is important to do this in
        // order to monitor the number of pending operations.
        final RequestContextImpl<R, ResultHandler<? super R>> requestContext =
            new RequestContextImpl<R, ResultHandler<? super R>>(
            this, resultHandler, messageID, false);
        if (addPendingRequest(requestContext))
        {
          // Find and cancel the request.
          final RequestContextImpl<?, ?> cancelledRequest = getPendingRequest(cancelRequest
              .getRequestID());
          if (cancelledRequest != null)
          {
            final LocalizableMessage cancelReason = INFO_CANCELED_BY_CANCEL_REQUEST
                .get(messageID);
            cancelledRequest
                .cancel(cancelReason, request, requestContext, true);
          }
          else
          {
            // Couldn't find the request. Invoke on context in order to remove
            // pending request.
            requestContext
                .handleErrorResult(newErrorResult(ResultCode.NO_SUCH_OPERATION));
          }
        }
      }
      else
      {
        final RequestContextImpl<R, ResultHandler<? super R>> requestContext;
        if (request.getOID().equals(StartTLSExtendedRequest.OID))
        {
          // StartTLS requests cannot be cancelled.
          requestContext = new RequestContextImpl<R, ResultHandler<? super R>>(
              this, resultHandler, messageID, false);
        }
        else
        {
          requestContext = new RequestContextImpl<R, ResultHandler<? super R>>(
              this, resultHandler, messageID, true);
        }

        if (addPendingRequest(requestContext))
        {
          requestHandler.handleExtendedRequest(requestContext, request,
              intermediateResponseHandler, requestContext);
        }
      }
    }



    /**
     * {@inheritDoc}
     */
    @Override
    public void handleModify(final Integer messageID,
        final ModifyRequest request,
        final IntermediateResponseHandler intermediateResponseHandler,
        final ResultHandler<? super Result> resultHandler)
    {
      final RequestContextImpl<Result, ResultHandler<? super Result>> requestContext =
          new RequestContextImpl<Result, ResultHandler<? super Result>>(
          this, resultHandler, messageID, true);
      if (addPendingRequest(requestContext))
      {
        requestHandler.handleModify(requestContext, request,
            intermediateResponseHandler, requestContext);
      }
    }



    /**
     * {@inheritDoc}
     */
    @Override
    public void handleModifyDN(final Integer messageID,
        final ModifyDNRequest request,
        final IntermediateResponseHandler intermediateResponseHandler,
        final ResultHandler<? super Result> resultHandler)
    {
      final RequestContextImpl<Result, ResultHandler<? super Result>> requestContext =
          new RequestContextImpl<Result, ResultHandler<? super Result>>(
          this, resultHandler, messageID, true);
      if (addPendingRequest(requestContext))
      {
        requestHandler.handleModifyDN(requestContext, request,
            intermediateResponseHandler, requestContext);
      }
    }



    /**
     * {@inheritDoc}
     */
    @Override
    public void handleSearch(final Integer messageID,
        final SearchRequest request,
        final IntermediateResponseHandler intermediateResponseHandler,
        final SearchResultHandler resultHandler)
    {
      final SearchRequestContextImpl requestContext = new SearchRequestContextImpl(
          this, resultHandler, messageID, true);
      if (addPendingRequest(requestContext))
      {
        requestHandler.handleSearch(requestContext, request,
            intermediateResponseHandler, requestContext);
      }
    }



    private boolean addPendingRequest(
        final RequestContextImpl<?, ?> requestContext)
    {
      final Integer messageID = requestContext.getMessageID();

      if (isClosed.get())
      {
        final LocalizableMessage message = INFO_CLIENT_CONNECTION_CLOSING.get();
        requestContext.handleErrorResult(newErrorResult(
            ResultCode.UNWILLING_TO_PERFORM, message.toString()));
        return false;
      }
      else if (pendingRequests.putIfAbsent(messageID, requestContext) != null)
      {
        final LocalizableMessage message = WARN_CLIENT_DUPLICATE_MESSAGE_ID
            .get(requestContext.getMessageID());
        requestContext.handleErrorResult(newErrorResult(
            ResultCode.PROTOCOL_ERROR, message.toString()));
        return false;
      }
      else if (isClosed.get())
      {
        // A concurrent close may have already removed the pending request but
        // it will have only been notified for cancellation.
        pendingRequests.remove(messageID);

        final LocalizableMessage message = INFO_CLIENT_CONNECTION_CLOSING.get();
        requestContext.handleErrorResult(newErrorResult(
            ResultCode.UNWILLING_TO_PERFORM, message.toString()));
        return false;
      }
      else
      {
        // If the connection is closed now then we just have to pay the cost of
        // invoking the request in the request handler.
        return true;
      }
    }



    private void doClose(final LocalizableMessage cancelReason)
    {
      if (!isClosed.getAndSet(true))
      {
        // At this point if any pending requests are added then we may end up
        // cancelling them, but this does not matter since addPendingRequest
        // will fail the request immediately.
        final Iterator<RequestContextImpl<?, ?>> iterator = pendingRequests
            .values().iterator();
        while (iterator.hasNext())
        {
          final RequestContextImpl<?, ?> pendingRequest = iterator.next();
          pendingRequest.cancel(cancelReason, null, null, false);
          iterator.remove();
        }
      }
    }



    /**
     * Returns the pending request context having the specified message ID.
     *
     * @param messageID
     *          The message ID associated with the request context.
     * @return The pending request context.
     */
    private RequestContextImpl<?, ?> getPendingRequest(final Integer messageID)
    {
      return pendingRequests.get(messageID);
    }



    /**
     * Deregister a request context once it has completed.
     *
     * @param requestContext
     *          The request context.
     * @return {@code true} if the request context was found and removed.
     */
    private boolean removePendingRequest(
        final RequestContextImpl<?, ?> requestContext)
    {
      return pendingRequests.remove(requestContext.getMessageID()) != null;
    }

  }



  private final RequestHandlerFactory<C, RequestContext> factory;



  /**
   * Creates a new server connection factory using the provided request handler
   * factory.
   *
   * @param factory
   *          The request handler factory to be adapted into a server connection
   *          factory.
   */
  RequestHandlerFactoryAdapter(
      final RequestHandlerFactory<C, RequestContext> factory)
  {
    this.factory = factory;
  }



  /**
   * {@inheritDoc}
   */
  @Override
  public ServerConnection<Integer> handleAccept(final C clientContext)
      throws ErrorResultException
  {
    final RequestHandler<RequestContext> requestHandler = factory
        .handleAccept(clientContext);
    return new ServerConnectionImpl<C>(requestHandler);
  }

}