/*
 * 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 2010 Sun Microsystems, Inc.
 *      Portions copyright 2011-2012 ForgeRock AS.
 */

package org.forgerock.opendj.ldap;

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

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.logging.Level;

import com.forgerock.opendj.util.AsynchronousFutureResult;
import com.forgerock.opendj.util.StaticUtils;
import com.forgerock.opendj.util.Validator;

/**
 * An abstract load balancing algorithm providing monitoring and failover
 * capabilities.
 * <p>
 * Implementations should override the method
 * {@code getInitialConnectionFactoryIndex()} in order to provide the policy for
 * selecting the first connection factory to use for each connection request.
 */
abstract class AbstractLoadBalancingAlgorithm implements LoadBalancingAlgorithm {
    private final class MonitoredConnectionFactory implements ConnectionFactory,
            ResultHandler<Connection> {

        private final ConnectionFactory factory;

        private final AtomicBoolean isOperational = new AtomicBoolean(true);

        private volatile FutureResult<?> pendingConnectFuture = null;

        private final int index;

        private MonitoredConnectionFactory(final ConnectionFactory factory, final int index) {
            this.factory = factory;
            this.index = index;
        }

        /**
         * {@inheritDoc}
         */
        public Connection getConnection() throws ErrorResultException {
            final Connection connection;
            try {
                connection = factory.getConnection();
            } catch (ErrorResultException e) {
                // Attempt failed - try next factory.
                notifyOffline(e);
                final int nextIndex = (index + 1) % monitoredFactories.size();
                final MonitoredConnectionFactory nextFactory =
                        getMonitoredConnectionFactory(nextIndex);
                return nextFactory.getConnection();
            }
            notifyOnline();
            return connection;
        }

        /**
         * {@inheritDoc}
         */
        @Override
        public FutureResult<Connection> getConnectionAsync(
                final ResultHandler<? super Connection> resultHandler) {
            final AsynchronousFutureResult<Connection> future =
                    new AsynchronousFutureResult<Connection>(resultHandler);

            final ResultHandler<Connection> failoverHandler = new ResultHandler<Connection>() {
                @Override
                public void handleErrorResult(final ErrorResultException error) {
                    // Attempt failed - try next factory.
                    notifyOffline(error);
                    final int nextIndex = (index + 1) % monitoredFactories.size();
                    try {
                        final MonitoredConnectionFactory nextFactory =
                                getMonitoredConnectionFactory(nextIndex);
                        nextFactory.getConnectionAsync(future);
                    } catch (final ErrorResultException e) {
                        future.handleErrorResult(e);
                    }
                }

                @Override
                public void handleResult(final Connection result) {
                    notifyOnline();
                    future.handleResult(result);
                }
            };

            factory.getConnectionAsync(failoverHandler);
            return future;
        }

        /**
         * Handle monitoring connection request failure.
         */
        @Override
        public void handleErrorResult(final ErrorResultException error) {
            notifyOffline(error);
        }

        /**
         * Handle monitoring connection request success.
         */
        @Override
        public void handleResult(final Connection connection) {
            notifyOnline();

            // The connection is not going to be used, so close it immediately.
            connection.close();
        }

        /**
         * {@inheritDoc}
         */
        @Override
        public String toString() {
            return factory.toString();
        }

        /**
         * Attempt to connect to the factory if it is offline and there is no
         * pending monitoring request.
         */
        private synchronized void checkIfAvailable() {
            if (!isOperational.get()
                    && (pendingConnectFuture == null || pendingConnectFuture.isDone())) {
                if (StaticUtils.DEBUG_LOG.isLoggable(Level.FINE)) {
                    StaticUtils.DEBUG_LOG.fine(String
                            .format("Attempting reconnect to offline factory " + this));
                }
                pendingConnectFuture = factory.getConnectionAsync(this);
            }
        }

        private void notifyOffline(final ErrorResultException error) {
            if (isOperational.getAndSet(false)) {
                // Transition from online to offline.
                if (StaticUtils.DEBUG_LOG.isLoggable(Level.WARNING)) {
                    StaticUtils.DEBUG_LOG.warning(String.format("Connection factory " + factory
                            + " is no longer operational: " + error.getMessage()));
                }

                synchronized (stateLock) {
                    offlineFactoriesCount++;
                    if (offlineFactoriesCount == 1) {
                        // Enable monitoring.
                        if (StaticUtils.DEBUG_LOG.isLoggable(Level.FINE)) {
                            StaticUtils.DEBUG_LOG.fine(String.format("Starting monitoring thread"));
                        }

                        monitoringFuture =
                                scheduler.scheduleWithFixedDelay(new MonitorRunnable(), 0,
                                        monitoringInterval, monitoringIntervalTimeUnit);
                    }
                }
            }
        }

        private void notifyOnline() {
            if (!isOperational.getAndSet(true)) {
                // Transition from offline to online.
                if (StaticUtils.DEBUG_LOG.isLoggable(Level.INFO)) {
                    StaticUtils.DEBUG_LOG.info(String.format("Connection factory " + factory
                            + " is now operational"));
                }

                synchronized (stateLock) {
                    offlineFactoriesCount--;
                    if (offlineFactoriesCount == 0) {
                        if (StaticUtils.DEBUG_LOG.isLoggable(Level.FINE)) {
                            StaticUtils.DEBUG_LOG.fine(String.format("Stopping monitoring thread"));
                        }

                        monitoringFuture.cancel(false);
                        monitoringFuture = null;
                    }
                }
            }
        }
    }

    private final class MonitorRunnable implements Runnable {
        private MonitorRunnable() {
            // Nothing to do.
        }

        @Override
        public void run() {
            for (final MonitoredConnectionFactory factory : monitoredFactories) {
                factory.checkIfAvailable();
            }
        }
    }

    private final List<MonitoredConnectionFactory> monitoredFactories;

    private final ScheduledExecutorService scheduler;

    private final Object stateLock = new Object();

    // Guarded by stateLock.
    private int offlineFactoriesCount = 0;

    private final long monitoringInterval;

    private final TimeUnit monitoringIntervalTimeUnit;

    // Guarded by stateLock.
    private ScheduledFuture<?> monitoringFuture;

    /**
     * Creates a new abstract load balancing algorithm which will monitor
     * offline connection factories every second using the default scheduler.
     *
     * @param factories
     *            The connection factories.
     */
    AbstractLoadBalancingAlgorithm(final Collection<ConnectionFactory> factories) {
        this(factories, 1, TimeUnit.SECONDS, StaticUtils.getDefaultScheduler());
    }

    /**
     * Creates a new abstract load balancing algorithm which will monitor
     * offline connection factories using the specified frequency using the
     * default scheduler.
     *
     * @param factories
     *            The connection factories.
     * @param interval
     *            The interval between attempts to poll offline factories.
     * @param unit
     *            The time unit for the interval between attempts to poll
     *            offline factories.
     */
    AbstractLoadBalancingAlgorithm(final Collection<ConnectionFactory> factories,
            final long interval, final TimeUnit unit) {
        this(factories, interval, unit, StaticUtils.getDefaultScheduler());
    }

    /**
     * Creates a new abstract load balancing algorithm which will monitor
     * offline connection factories using the specified frequency and scheduler.
     *
     * @param factories
     *            The connection factories.
     * @param interval
     *            The interval between attempts to poll offline factories.
     * @param unit
     *            The time unit for the interval between attempts to poll
     *            offline factories.
     * @param scheduler
     *            The scheduler which should for periodically monitoring dead
     *            connection factories to see if they are usable again.
     */
    AbstractLoadBalancingAlgorithm(final Collection<ConnectionFactory> factories,
            final long interval, final TimeUnit unit, final ScheduledExecutorService scheduler) {
        Validator.ensureNotNull(factories, scheduler, unit);

        this.monitoredFactories = new ArrayList<MonitoredConnectionFactory>(factories.size());
        int i = 0;
        for (final ConnectionFactory f : factories) {
            this.monitoredFactories.add(new MonitoredConnectionFactory(f, i++));
        }
        this.scheduler = scheduler;
        this.monitoringInterval = interval;
        this.monitoringIntervalTimeUnit = unit;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public final ConnectionFactory getConnectionFactory() throws ErrorResultException {
        final int index = getInitialConnectionFactoryIndex();
        return getMonitoredConnectionFactory(index);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public String toString() {
        final StringBuilder builder = new StringBuilder();
        builder.append(getAlgorithmName());
        builder.append('(');
        boolean isFirst = true;
        for (final ConnectionFactory factory : monitoredFactories) {
            if (!isFirst) {
                builder.append(',');
            } else {
                isFirst = false;
            }
            builder.append(factory);
        }
        builder.append(')');
        return builder.toString();
    }

    /**
     * Returns the name of this load balancing algorithm.
     *
     * @return The name of this load balancing algorithm.
     */
    abstract String getAlgorithmName();

    /**
     * Returns the index of the first connection factory which should be used in
     * order to satisfy the next connection request.
     *
     * @return The index of the first connection factory which should be used in
     *         order to satisfy the next connection request.
     */
    abstract int getInitialConnectionFactoryIndex();

    // Return the first factory after index which is operational.
    private MonitoredConnectionFactory getMonitoredConnectionFactory(final int initialIndex)
            throws ErrorResultException {
        int index = initialIndex;
        final int maxIndex = monitoredFactories.size();
        do {
            final MonitoredConnectionFactory factory = monitoredFactories.get(index);
            if (factory.isOperational.get()) {
                return factory;
            }
            index = (index + 1) % maxIndex;
        } while (index != initialIndex);

        // All factories are offline so give up. We could have a
        // configurable policy here such as waiting indefinitely, or for a
        // configurable timeout period.
        throw newErrorResult(ResultCode.CLIENT_SIDE_CONNECT_ERROR,
                "No operational connection factories available");
    }
}