/*
 * 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 2015-2016 ForgeRock AS.
 */
package org.forgerock.util.promise;

import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import org.forgerock.util.AsyncFunction;
import org.forgerock.util.Function;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * An implementation of {@link Promise} which can be used as is, or as the basis
 * for more complex asynchronous behavior. A {@code PromiseImpl} must be
 * completed by invoking one of:
 * <ul>
 * <li>{@link #handleResult} - marks the promise as having succeeded with the
 * provide result
 * <li>{@link #handleException} - marks the promise as having failed with the
 * provided exception
 * <li>{@link #cancel} - requests cancellation of the asynchronous task
 * represented by the promise. Cancellation is only supported if the
 * {@link #tryCancel(boolean)} is overridden and returns an exception.
 * </ul>
 *
 * @param <V>
 *            The type of the task's result, or {@link Void} if the task does
 *            not return anything (i.e. it only has side-effects).
 * @param <E>
 *            The type of the exception thrown by the task if it fails, or
 *            {@link NeverThrowsException} if the task cannot fail.
 * @see Promise
 * @see Promises
 */
public class PromiseImpl<V, E extends Exception> implements Promise<V, E>, ResultHandler<V>,
        ExceptionHandler<E>, RuntimeExceptionHandler {
    // TODO: Is using monitor based sync better than AQS?

    private static final Logger LOGGER = LoggerFactory.getLogger(PromiseImpl.class);

    private interface StateListener<V, E extends Exception> {
        void handleStateChange(int newState, V result, E exception, RuntimeException runtimeException);
    }

    /**
     * State value indicating that this promise has not completed.
     */
    private static final int PENDING = 0;

    /**
     * State value indicating that this promise has completed successfully
     * (result set).
     */
    private static final int HAS_RESULT = 1;

    /**
     * State value indicating that this promise has failed (exception set).
     */
    private static final int HAS_EXCEPTION = 2;

    /**
     * State value indicating that this promise has been cancelled (exception set).
     */
    private static final int CANCELLED = 3;

    /**
     * State value indicating that this promise has failed with a runtime exception.
     */
    private static final int HAS_RUNTIME_EXCEPTION = 4;

    /**
     * Creates a new pending {@link Promise} implementation.
     *
     * @param <V>
     *            The type of the task's result, or {@link Void} if the task
     *            does not return anything (i.e. it only has side-effects).
     * @param <E>
     *            The type of the exception thrown by the task if it fails, or
     *            {@link NeverThrowsException} if the task cannot fail.
     * @return A new pending {@link Promise} implementation.
     */
    public static <V, E extends Exception> PromiseImpl<V, E> create() {
        return new PromiseImpl<>();
    }

    private volatile int state = PENDING;
    private V result = null;
    private E exception = null;
    private RuntimeException runtimeException = null;

    private final Queue<StateListener<V, E>> listeners =
            new ConcurrentLinkedQueue<>();

    /**
     * Creates a new pending {@link Promise} implementation. This constructor is
     * protected to allow for sub-classing.
     */
    protected PromiseImpl() {
        // No implementation.
    }

    @Override
    public final boolean cancel(final boolean mayInterruptIfRunning) {
        if (isDone()) {
            // Fail-fast.
            return false;
        }
        final E exception = tryCancel(mayInterruptIfRunning);
        return exception != null && setState(CANCELLED, null, exception, null);
    }

    @Override
    public final V get() throws InterruptedException, ExecutionException {
        await(); // Publishes.
        return get0();
    }

    @Override
    public final V get(final long timeout, final TimeUnit unit) throws InterruptedException,
            ExecutionException, TimeoutException {
        await(timeout, unit, false); // Publishes.
        return get0();
    }

    @Override
    public final V getOrThrow() throws InterruptedException, E {
        await(); // Publishes.
        return getOrThrow0();
    }

    @Override
    public final V getOrThrow(final long timeout, final TimeUnit unit) throws InterruptedException,
            E, TimeoutException {
        await(timeout, unit, false); // Publishes.
        return getOrThrow0();
    }

    @Override
    public final V getOrThrowUninterruptibly() throws E {
        boolean wasInterrupted = false;
        try {
            while (true) {
                try {
                    return getOrThrow();
                } catch (final InterruptedException e) {
                    wasInterrupted = true;
                }
            }
        } finally {
            if (wasInterrupted) {
                Thread.currentThread().interrupt();
            }
        }
    }

    @Override
    public final V getOrThrowUninterruptibly(final long timeout, final TimeUnit unit) throws E,
            TimeoutException {
        try {
            await(timeout, unit, true); // Publishes.
        } catch (InterruptedException ignored) {
            // Will never occur since interrupts are ignored.
        }
        return getOrThrow0();
    }

    /**
     * Signals that the asynchronous task represented by this promise has
     * failed. If the task has already completed (i.e. {@code isDone() == true})
     * then calling this method has no effect and the provided result will be
     * discarded.
     *
     * @param exception
     *            The exception indicating why the task failed.
     * @see #tryHandleException(Exception)
     */
    @Override
    public final void handleException(final E exception) {
        tryHandleException(exception);
    }

    @Override
    public void handleRuntimeException(RuntimeException exception) {
        setState(HAS_RUNTIME_EXCEPTION, null, null, exception);
    }

    /**
     * Signals that the asynchronous task represented by this promise has
     * succeeded. If the task has already completed (i.e.
     * {@code isDone() == true}) then calling this method has no effect and the
     * provided result will be discarded.
     *
     * @param result
     *            The result of the asynchronous task (may be {@code null}).
     * @see #tryHandleResult(Object)
     */
    @Override
    public final void handleResult(final V result) {
        tryHandleResult(result);
    }

    /**
     * Attempts to signal that the asynchronous task represented by this promise
     * has failed. If the task has already completed (i.e.
     * {@code isDone() == true}) then calling this method has no effect and
     * {@code false} is returned.
     * <p>
     * This method should be used in cases where multiple threads may
     * concurrently attempt to complete a promise and need to release resources
     * if the completion attempt fails. For example, an asynchronous TCP connect
     * attempt may complete after a timeout has expired. In this case the
     * connection should be immediately closed because it is never going to be
     * used.
     *
     * @param exception
     *            The exception indicating why the task failed.
     * @return {@code false} if this promise has already been completed, either
     *         due to normal termination, an exception, or cancellation (i.e.
     *         {@code isDone() == true}).
     * @see #handleException(Exception)
     * @see #isDone()
     */
    public final boolean tryHandleException(final E exception) {
        return setState(HAS_EXCEPTION, null, exception, null);
    }

    /**
     * Attempts to signal that the asynchronous task represented by this promise
     * has succeeded. If the task has already completed (i.e.
     * {@code isDone() == true}) then calling this method has no effect and
     * {@code false} is returned.
     * <p>
     * This method should be used in cases where multiple threads may
     * concurrently attempt to complete a promise and need to release resources
     * if the completion attempt fails. For example, an asynchronous TCP connect
     * attempt may complete after a timeout has expired. In this case the
     * connection should be immediately closed because it is never going to be
     * used.
     *
     * @param result
     *            The result of the asynchronous task (may be {@code null}).
     * @return {@code false} if this promise has already been completed, either
     *         due to normal termination, an exception, or cancellation (i.e.
     *         {@code isDone() == true}).
     * @see #handleResult(Object)
     * @see #isDone()
     */
    public final boolean tryHandleResult(final V result) {
        return setState(HAS_RESULT, result, null, null);
    }

    @Override
    public final boolean isCancelled() {
        return state == CANCELLED;
    }

    @Override
    public final boolean isDone() {
        return state != PENDING;
    }

    @Override
    public final Promise<V, E> thenOnException(final ExceptionHandler<? super E> onException) {
        addOrFireListener(new StateListener<V, E>() {
            @Override
            public void handleStateChange(final int newState, final V result, final E exception,
                    final RuntimeException runtimeException) {
                if (newState == HAS_EXCEPTION || newState == CANCELLED) {
                    try {
                        onException.handleException(exception);
                    } catch (RuntimeException e) {
                        LOGGER.error("Ignored unexpected exception thrown by ExceptionHandler", e);
                    }
                }
            }
        });
        return this;
    }

    @Override
    public final Promise<V, E> thenOnResult(final ResultHandler<? super V> onResult) {
        addOrFireListener(new StateListener<V, E>() {
            @Override
            public void handleStateChange(final int newState, final V result, final E exception,
                    final RuntimeException runtimeException) {
                if (newState == HAS_RESULT) {
                    try {
                        onResult.handleResult(result);
                    } catch (RuntimeException e) {
                        LOGGER.error("Ignored unexpected exception thrown by ResultHandler", e);
                    }
                }
            }
        });
        return this;
    }

    @Override
    public final Promise<V, E> thenOnResultOrException(final ResultHandler<? super V> onResult,
            final ExceptionHandler<? super E> onException) {
        addOrFireListener(new StateListener<V, E>() {
            @Override
            public void handleStateChange(final int newState, final V result, final E exception,
                    final RuntimeException runtimeException) {
                if (newState == HAS_RESULT) {
                    try {
                        onResult.handleResult(result);
                    } catch (RuntimeException e) {
                        LOGGER.error("Ignored unexpected exception thrown by ResultHandler", e);
                    }
                } else if (newState == HAS_EXCEPTION || newState == CANCELLED) {
                    try {
                        onException.handleException(exception);
                    } catch (RuntimeException e) {
                        LOGGER.error("Ignored unexpected exception thrown by ExceptionHandler", e);
                    }
                }
            }
        });
        return this;
    }

    @Override
    public final Promise<V, E> thenOnResultOrException(final Runnable onResultOrException) {
        addOrFireListener(new StateListener<V, E>() {
            @Override
            public void handleStateChange(final int newState, final V result, final E exception,
                    final RuntimeException runtimeException) {
                if (newState != HAS_RUNTIME_EXCEPTION) {
                    try {
                        onResultOrException.run();
                    } catch (RuntimeException e) {
                        LOGGER.error("Ignored unexpected exception thrown by Runnable", e);
                    }
                }
            }
        });
        return this;
    }

    @Override
    public final <VOUT> Promise<VOUT, E> then(final Function<? super V, VOUT, E> onResult) {
        return then(onResult, Promises.<VOUT, E>exceptionIdempotentFunction());
    }

    @Override
    public <EOUT extends Exception> Promise<V, EOUT> thenCatch(final Function<? super E, V, EOUT> onException) {
        return then(Promises.<V, EOUT>resultIdempotentFunction(), onException);
    }

    @Override
    public Promise<V, E> thenCatchRuntimeException(
            Function<? super RuntimeException, V, E> onRuntimeException) {
        return then(Promises.<V, E>resultIdempotentFunction(), Promises.<V, E>exceptionIdempotentFunction(),
                onRuntimeException);
    }

    @Override
    public final <VOUT, EOUT extends Exception> Promise<VOUT, EOUT> then(
            final Function<? super V, VOUT, EOUT> onResult, final Function<? super E, VOUT, EOUT> onException) {
        return then(onResult, onException, Promises.<VOUT, EOUT>runtimeExceptionIdempotentFunction());
    }

    @Override
    @SuppressWarnings("unchecked")
    public final <VOUT, EOUT extends Exception> Promise<VOUT, EOUT> then(
            final Function<? super V, VOUT, EOUT> onResult, final Function<? super E, VOUT, EOUT> onException,
            final Function<? super RuntimeException, VOUT, EOUT> onRuntimeException) {
        final PromiseImpl<VOUT, EOUT> chained = new PromiseImpl<>();
        addOrFireListener(new StateListener<V, E>() {
            @Override
            public void handleStateChange(final int newState, final V result, final E exception,
                                          final RuntimeException runtimeException) {
                try {
                    switch (newState) {
                        case HAS_RESULT:
                            chained.handleResult(onResult.apply(result));
                            break;
                        case HAS_EXCEPTION:
                        case CANCELLED:
                            chained.handleResult(onException.apply(exception));
                            break;
                        case HAS_RUNTIME_EXCEPTION:
                            chained.handleResult(onRuntimeException.apply(runtimeException));
                            break;
                        default:
                            throw new IllegalStateException("Unexpected state : " + newState);
                    }
                } catch (final RuntimeException e) {
                    tryHandlingRuntimeException(e, chained);
                } catch (final Exception e) {
                    chained.handleException((EOUT) e);
                }
            }
        });
        return chained;
    }

    private <VOUT, EOUT extends Exception> void tryHandlingRuntimeException(final RuntimeException runtimeException,
            final PromiseImpl<VOUT, EOUT> chained) {
        try {
            chained.handleRuntimeException(runtimeException);
        } catch (Exception ignored) {
            LOGGER.error("Runtime exception handler threw a RuntimeException which cannot be handled!", ignored);
        }
    }

    @Override
    public final Promise<V, E> thenAlways(final Runnable always) {
        addOrFireListener(new StateListener<V, E>() {
            @Override
            public void handleStateChange(final int newState, final V result, final E exception,
                    final RuntimeException runtimeException) {
                try {
                    always.run();
                } catch (RuntimeException e) {
                    LOGGER.error("Ignored unexpected exception thrown by Runnable", e);
                }
            }
        });
        return this;
    }

    @Override
    public final Promise<V, E> thenFinally(final Runnable onFinally) {
        return thenAlways(onFinally);
    }

    @Override
    public final <VOUT> Promise<VOUT, E> thenAsync(final AsyncFunction<? super V, VOUT, E> onResult) {
        return thenAsync(onResult, Promises.<VOUT, E>exceptionIdempotentAsyncFunction());
    }

    @Override
    public final <EOUT extends Exception> Promise<V, EOUT> thenCatchAsync(
            AsyncFunction<? super E, V, EOUT> onException) {
        return thenAsync(Promises.<V, EOUT>resultIdempotentAsyncFunction(), onException);
    }

    @Override
    public final Promise<V, E> thenCatchRuntimeExceptionAsync(
            AsyncFunction<? super RuntimeException, V, E> onRuntimeException) {
        return thenAsync(Promises.<V, E>resultIdempotentAsyncFunction(),
                Promises.<V, E>exceptionIdempotentAsyncFunction(), onRuntimeException);
    }

    @Override
    public final <VOUT, EOUT extends Exception> Promise<VOUT, EOUT> thenAsync(
            final AsyncFunction<? super V, VOUT, EOUT> onResult,
            final AsyncFunction<? super E, VOUT, EOUT> onException) {
        return thenAsync(onResult, onException, Promises.<VOUT, EOUT>runtimeExceptionIdempotentAsyncFunction());
    }

    @Override
    public final <VOUT, EOUT extends Exception> Promise<VOUT, EOUT> thenAsync(
            final AsyncFunction<? super V, VOUT, EOUT> onResult,
            final AsyncFunction<? super E, VOUT, EOUT> onException,
            final AsyncFunction<? super RuntimeException, VOUT, EOUT> onRuntimeException) {
        final PromiseImpl<VOUT, EOUT> chained = new PromiseImpl<>();
        addOrFireListener(new StateListener<V, E>() {
            @Override
            @SuppressWarnings("unchecked")
            public void handleStateChange(final int newState, final V result, final E exception,
                    final RuntimeException runtimeException) {
                try {
                    switch (newState) {
                        case HAS_RESULT:
                            callNestedPromise(onResult.apply(result));
                            break;
                        case HAS_EXCEPTION:
                        case CANCELLED:
                            callNestedPromise(onException.apply(exception));
                            break;
                        case HAS_RUNTIME_EXCEPTION:
                            callNestedPromise(onRuntimeException.apply(runtimeException));
                            break;
                        default:
                            throw new IllegalStateException("Unexpected state : " + newState);
                    }
                } catch (final RuntimeException e) {
                    tryHandlingRuntimeException(e, chained);
                } catch (final Exception e) {
                    chained.handleException((EOUT) e);
                }
            }

            private void callNestedPromise(Promise<? extends VOUT, ? extends EOUT> nestedPromise) {
                nestedPromise
                        .thenOnResult(chained)
                        .thenOnException(chained)
                        .thenOnRuntimeException(chained);
            }
        });
        return chained;
    }

    @Override
    public final Promise<V, E> thenOnRuntimeException(final RuntimeExceptionHandler onRuntimeException) {
        addOrFireListener(new StateListener<V, E>() {
            @Override
            public void handleStateChange(int newState, V result, E exception, RuntimeException runtimeException) {
                if (newState == HAS_RUNTIME_EXCEPTION) {
                    try {
                        onRuntimeException.handleRuntimeException(runtimeException);
                    } catch (RuntimeException e) {
                        LOGGER.error("Ignored unexpected exception thrown by RuntimeExceptionHandler", e);
                    }
                }
            }
        });
        return this;
    }

    /**
     * Invoked when the client attempts to cancel the asynchronous task
     * represented by this promise. Implementations which support cancellation
     * should override this method to cancel the asynchronous task and, if
     * successful, return an appropriate exception which can be used to signal
     * that the task has failed.
     * <p>
     * By default cancellation is not supported and this method returns
     * {@code null}.
     *
     * @param mayInterruptIfRunning
     *            {@code true} if the thread executing this task should be
     *            interrupted; otherwise, in-progress tasks are allowed to
     *            complete.
     * @return {@code null} if cancellation was not supported or not possible,
     *         otherwise an appropriate exception.
     */
    protected E tryCancel(final boolean mayInterruptIfRunning) {
        return null;
    }

    private void addOrFireListener(final StateListener<V, E> listener) {
        final int stateBefore = state;
        if (stateBefore != PENDING) {
            handleCompletion(listener, stateBefore);
        } else {
            listeners.add(listener);
            final int stateAfter = state;
            if (stateAfter != PENDING && listeners.remove(listener)) {
                handleCompletion(listener, stateAfter);
            }
        }
    }

    private void handleCompletion(final StateListener<V, E> listener, final int completedState) {
        try {
            listener.handleStateChange(completedState, result, exception, runtimeException);
        } catch (RuntimeException ignored) {
            LOGGER.error("State change listener threw a RuntimeException which cannot be handled!", ignored);
        }
    }

    private V get0() throws ExecutionException {
        if (runtimeException != null) {
            throw new ExecutionException(runtimeException);
        } else if (exception != null) {
            throw new ExecutionException(exception);
        } else {
            return result;
        }
    }

    private V getOrThrow0() throws E {
        if (runtimeException != null) {
            throw runtimeException;
        } else if (exception != null) {
            throw exception;
        } else {
            return result;
        }
    }

    private boolean setState(final int newState, final V result, final E exception,
            final RuntimeException runtimeException) {
        synchronized (this) {
            if (state != PENDING) {
                // Already completed.
                return false;
            }
            this.result = result;
            this.exception = exception;
            this.runtimeException = runtimeException;
            state = newState; // Publishes.
            notifyAll(); // Wake up any blocked threads.
        }
        StateListener<V, E> listener;
        while ((listener = listeners.poll()) != null) {
            handleCompletion(listener, newState);
        }
        return true;
    }

    private void await() throws InterruptedException {
        // Use double-check for fast-path.
        if (state == PENDING) {
            synchronized (this) {
                while (state == PENDING) {
                    wait();
                }
            }
        }
    }

    private void await(final long timeout, final TimeUnit unit, final boolean isUninterruptibly)
            throws InterruptedException, TimeoutException {
        // Use double-check for fast-path.
        if (state == PENDING) {
            final long timeoutMS = unit.toMillis(timeout);
            final long endTimeMS = System.currentTimeMillis() + timeoutMS;
            boolean wasInterrupted = false;
            try {
                synchronized (this) {
                    while (state == PENDING) {
                        final long remainingTimeMS = endTimeMS - System.currentTimeMillis();
                        if (remainingTimeMS <= 0) {
                            throw new TimeoutException();
                        }
                        try {
                            wait(remainingTimeMS);
                        } catch (final InterruptedException e) {
                            if (isUninterruptibly) {
                                wasInterrupted = true;
                            } else {
                                throw e;
                            }
                        }
                    }
                }
            } finally {
                if (wasInterrupted) {
                    Thread.currentThread().interrupt();
                }
            }
        }
    }
}