/*
 * 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-2013 ForgeRock AS.
 */

package org.forgerock.opendj.ldap;

import static com.forgerock.opendj.util.StaticUtils.DEBUG_LOG;
import static com.forgerock.opendj.util.StaticUtils.DEFAULT_SCHEDULER;
import static org.forgerock.opendj.ldap.ErrorResultException.*;

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.ReferenceCountedObject;
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;
        }

        @Override
        public void close() {
            // Should we cancel the future?
            factory.close();
        }

        @Override
        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;
        }

        @Override
        public FutureResult<Connection> getConnectionAsync(
                final ResultHandler<? super Connection> resultHandler) {
            final AsynchronousFutureResult<Connection, ResultHandler<? super Connection>> future =
                    new AsynchronousFutureResult<Connection, ResultHandler<? super 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();
        }

        @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 (DEBUG_LOG.isLoggable(Level.FINE)) {
                    DEBUG_LOG.fine(String.format("Attempting reconnect to offline factory '%s'",
                            this));
                }
                pendingConnectFuture = factory.getConnectionAsync(this);
            }
        }

        private void notifyOffline(final ErrorResultException error) {
            // Save the error in case the load-balancer is exhausted.
            lastFailure = error;

            if (isOperational.getAndSet(false)) {
                // Transition from online to offline.
                synchronized (listenerLock) {
                    try {
                        listener.handleConnectionFactoryOffline(factory, error);
                    } catch (RuntimeException e) {
                        handleListenerException(e);
                    }
                }

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

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

        private void notifyOnline() {
            if (!isOperational.getAndSet(true)) {
                // Transition from offline to online.
                synchronized (listenerLock) {
                    try {
                        listener.handleConnectionFactoryOnline(factory);
                    } catch (RuntimeException e) {
                        handleListenerException(e);
                    }
                }

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

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

        private void handleListenerException(RuntimeException e) {
            if (DEBUG_LOG.isLoggable(Level.SEVERE)) {
                DEBUG_LOG.log(Level.SEVERE,
                        "A run-time error occurred while processing a load-balancer event", e);
            }
        }
    }

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

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

    /**
     * A default event listener which just logs the event.
     */
    private static final LoadBalancerEventListener DEFAULT_LISTENER =
            new LoadBalancerEventListener() {

                @Override
                public void handleConnectionFactoryOnline(ConnectionFactory factory) {
                    // Transition from offline to online.
                    if (DEBUG_LOG.isLoggable(Level.INFO)) {
                        DEBUG_LOG.info(String.format("Connection factory'%s' is now operational",
                                factory));
                    }
                }

                @Override
                public void handleConnectionFactoryOffline(ConnectionFactory factory,
                        ErrorResultException error) {
                    if (DEBUG_LOG.isLoggable(Level.WARNING)) {
                        DEBUG_LOG.warning(String.format(
                                "Connection factory '%s' is no longer operational: %s", factory,
                                error.getMessage()));
                    }
                }
            };

    private final List<MonitoredConnectionFactory> monitoredFactories;
    private final ReferenceCountedObject<ScheduledExecutorService>.Reference scheduler;
    private final Object stateLock = new Object();

    /**
     * The last connection failure which caused a connection factory to be
     * marked offline. This is used in order to help diagnose problems when the
     * load-balancer has exhausted all of its factories.
     */
    private volatile ErrorResultException lastFailure = null;

    /**
     * The event listener which should be notified when connection factories go
     * on or off-line.
     */
    private final LoadBalancerEventListener listener;

    /**
     * Ensures that events are notified one at a time.
     */
    private final Object listenerLock = new Object();

    /**
     * Guarded by stateLock.
     */
    private int offlineFactoriesCount = 0;
    private final long monitoringInterval;
    private final TimeUnit monitoringIntervalTimeUnit;
    /**
     * Guarded by stateLock.
     */
    private ScheduledFuture<?> monitoringFuture;
    private AtomicBoolean isClosed = new AtomicBoolean();

    AbstractLoadBalancingAlgorithm(final Collection<? extends ConnectionFactory> factories,
            final LoadBalancerEventListener listener, final long interval, final TimeUnit unit,
            final ScheduledExecutorService scheduler) {
        Validator.ensureNotNull(factories, 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 = DEFAULT_SCHEDULER.acquireIfNull(scheduler);
        this.monitoringInterval = interval;
        this.monitoringIntervalTimeUnit = unit;
        this.listener = listener != null ? listener : DEFAULT_LISTENER;
    }

    @Override
    public void close() {
        if (isClosed.compareAndSet(false, true)) {
            synchronized (stateLock) {
                if (monitoringFuture != null) {
                    monitoringFuture.cancel(false);
                    monitoringFuture = null;
                }
            }
            for (ConnectionFactory factory : monitoredFactories) {
                factory.close();
            }
            scheduler.release();
        }
    }

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

    @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", lastFailure);
    }
}