/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *      https://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.commons.lang3.concurrent;

import java.beans.PropertyChangeListener;
import java.util.EnumMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;

/**
 * A simple implementation of the <a
 * href="https://martinfowler.com/bliki/CircuitBreaker.html">Circuit Breaker</a> pattern
 * that counts specific events.
 *
 * <p>
 * A <em>circuit breaker</em> can be used to protect an application against unreliable
 * services or unexpected load. A newly created {@link EventCountCircuitBreaker} object is
 * initially in state <em>closed</em> meaning that no problem has been detected. When the
 * application encounters specific events (like errors or service timeouts), it tells the
 * circuit breaker to increment an internal counter. If the number of events reported in a
 * specific time interval exceeds a configurable threshold, the circuit breaker changes
 * into state <em>open</em>. This means that there is a problem with the associated sub
 * system; the application should no longer call it, but give it some time to settle down.
 * The circuit breaker can be configured to switch back to <em>closed</em> state after a
 * certain time frame if the number of events received goes below a threshold.
 * </p>
 * <p>
 * When a {@link EventCountCircuitBreaker} object is constructed the following parameters
 * can be provided:
 * </p>
 * <ul>
 * <li>A threshold for the number of events that causes a state transition to
 * <em>open</em> state. If more events are received in the configured check interval, the
 * circuit breaker switches to <em>open</em> state.</li>
 * <li>The interval for checks whether the circuit breaker should open. So it is possible
 * to specify something like "The circuit breaker should open if more than 10 errors are
 * encountered in a minute."</li>
 * <li>The same parameters can be specified for automatically closing the circuit breaker
 * again, as in "If the number of requests goes down to 100 per minute, the circuit
 * breaker should close itself again". Depending on the use case, it may make sense to use
 * a slightly lower threshold for closing the circuit breaker than for opening it to avoid
 * continuously flipping when the number of events received is close to the threshold.</li>
 * </ul>
 * <p>
 * This class supports the following typical use cases:
 * </p>
 * <p>
 * <strong>Protecting against load peaks</strong>
 * </p>
 * <p>
 * Imagine you have a server which can handle a certain number of requests per minute.
 * Suddenly, the number of requests increases significantly - maybe because a connected
 * partner system is going mad or due to a denial of service attack. A
 * {@link EventCountCircuitBreaker} can be configured to stop the application from
 * processing requests when a sudden peak load is detected and to start request processing
 * again when things calm down. The following code fragment shows a typical example of
 * such a scenario. Here the {@link EventCountCircuitBreaker} allows up to 1000 requests
 * per minute before it interferes. When the load goes down again to 800 requests per
 * second it switches back to state <em>closed</em>:
 * </p>
 *
 * <pre>
 * EventCountCircuitBreaker breaker = new EventCountCircuitBreaker(1000, 1, TimeUnit.MINUTE, 800);
 * ...
 * public void handleRequest(Request request) {
 *     if (breaker.incrementAndCheckState()) {
 *         // actually handle this request
 *     } else {
 *         // do something else, e.g. send an error code
 *     }
 * }
 * </pre>
 * <p>
 * <strong>Deal with an unreliable service</strong>
 * </p>
 * <p>
 * In this scenario, an application uses an external service which may fail from time to
 * time. If there are too many errors, the service is considered down and should not be
 * called for a while. This can be achieved using the following pattern - in this concrete
 * example we accept up to 5 errors in 2 minutes; if this limit is reached, the service is
 * given a rest time of 10 minutes:
 * </p>
 *
 * <pre>
 * EventCountCircuitBreaker breaker = new EventCountCircuitBreaker(5, 2, TimeUnit.MINUTE, 5, 10, TimeUnit.MINUTE);
 * ...
 * public void handleRequest(Request request) {
 *     if (breaker.checkState()) {
 *         try {
 *             service.doSomething();
 *         } catch (ServiceException ex) {
 *             breaker.incrementAndCheckState();
 *         }
 *     } else {
 *         // return an error code, use an alternative service, etc.
 *     }
 * }
 * </pre>
 * <p>
 * In addition to automatic state transitions, the state of a circuit breaker can be
 * changed manually using the methods {@link #open()} and {@link #close()}. It is also
 * possible to register {@link PropertyChangeListener} objects that get notified whenever
 * a state transition occurs. This is useful, for instance to directly react on a freshly
 * detected error condition.
 * </p>
 * <p>
 * <em>Implementation notes:</em>
 * </p>
 * <ul>
 * <li>This implementation uses non-blocking algorithms to update the internal counter and
 * state. This should be pretty efficient if there is not too much contention.</li>
 * <li>This implementation is not intended to operate as a high-precision timer in very
 * short check intervals. It is deliberately kept simple to avoid complex and
 * time-consuming state checks. It should work well in time intervals from a few seconds
 * up to minutes and longer. If the intervals become too short, there might be race
 * conditions causing spurious state transitions.</li>
 * <li>The handling of check intervals is a bit simplistic. Therefore, there is no
 * guarantee that the circuit breaker is triggered at a specific point in time; there may
 * be some delay (less than a check interval).</li>
 * </ul>
 *
 * @since 3.5
 */
public class EventCountCircuitBreaker extends AbstractCircuitBreaker<Integer> {

    /**
     * Internally used class for executing check logic based on the current state of the
     * circuit breaker. Having this logic extracted into special classes avoids complex
     * if-then-else cascades.
     */
    private abstract static class AbstractStateStrategy {

        /**
         * Obtains the check interval to applied for the represented state from the given
         * {@link CircuitBreaker}.
         *
         * @param breaker The {@link CircuitBreaker}
         * @return The check interval to be applied
         */
        protected abstract long fetchCheckInterval(EventCountCircuitBreaker breaker);

        /**
         * Tests whether the end of the current check interval is reached.
         *
         * @param breaker The {@link CircuitBreaker}
         * @param currentData The current state object
         * @param now The current time
         * @return A flag whether the end of the current check interval is reached
         */
        public boolean isCheckIntervalFinished(final EventCountCircuitBreaker breaker,
                final CheckIntervalData currentData, final long now) {
            return now - currentData.getCheckIntervalStart() > fetchCheckInterval(breaker);
        }

        /**
         * Tests whether the specified {@link CheckIntervalData} objects indicate that a
         * state transition should occur. Here the logic which checks for thresholds
         * depending on the current state is implemented.
         *
         * @param breaker The {@link CircuitBreaker}
         * @param currentData The current {@link CheckIntervalData} object
         * @param nextData The updated {@link CheckIntervalData} object
         * @return A flag whether a state transition should be performed
         */
        public abstract boolean isStateTransition(EventCountCircuitBreaker breaker,
                CheckIntervalData currentData, CheckIntervalData nextData);
    }

    /**
     * An internally used data class holding information about the checks performed by
     * this class. Basically, the number of received events and the start time of the
     * current check interval are stored.
     */
    private static final class CheckIntervalData {

        /** The counter for events. */
        private final long eventCount;

        /** The start time of the current check interval. */
        private final long checkIntervalStart;

        /**
         * Creates a new instance of {@link CheckIntervalData}.
         *
         * @param count The current count value
         * @param intervalStart The start time of the check interval
         */
        CheckIntervalData(final long count, final long intervalStart) {
            eventCount = count;
            checkIntervalStart = intervalStart;
        }

        /**
         * Gets the start time of the current check interval.
         *
         * @return The check interval start time
         */
        public long getCheckIntervalStart() {
            return checkIntervalStart;
        }

        /**
         * Gets the event counter.
         *
         * @return The number of received events
         */
        public long getEventCount() {
            return eventCount;
        }

        /**
         * Returns a new instance of {@link CheckIntervalData} with the event counter incremented by the given delta. If the delta is 0, this object is
         * returned.
         * <p>
         * If the increment would overflow, the event counter saturates at {@link Long#MAX_VALUE}.
         * </p>
         *
         * @param delta The delta
         * @return The updated instance
         */
        public CheckIntervalData increment(final int delta) {
            if (delta == 0) {
                return this;
            }
            final long nextCount = getEventCount() + delta;
            return new CheckIntervalData(nextCount < getEventCount() ? Long.MAX_VALUE : nextCount, getCheckIntervalStart());
        }
    }

    /**
     * A specialized {@link AbstractStateStrategy} implementation for the state closed.
     */
    private static final class StateStrategyClosed extends AbstractStateStrategy {

        /**
         * {@inheritDoc}
         */
        @Override
        protected long fetchCheckInterval(final EventCountCircuitBreaker breaker) {
            return breaker.getOpeningInterval();
        }

        /**
         * {@inheritDoc}
         */
        @Override
        public boolean isStateTransition(final EventCountCircuitBreaker breaker,
                final CheckIntervalData currentData, final CheckIntervalData nextData) {
            return nextData.getEventCount() > breaker.getOpeningThreshold();
        }
    }

    /**
     * A specialized {@link AbstractStateStrategy} implementation for the state open.
     */
    private static final class StateStrategyOpen extends AbstractStateStrategy {

        /**
         * {@inheritDoc}
         */
        @Override
        protected long fetchCheckInterval(final EventCountCircuitBreaker breaker) {
            return breaker.getClosingInterval();
        }

        /**
         * {@inheritDoc}
         */
        @Override
        public boolean isStateTransition(final EventCountCircuitBreaker breaker,
                final CheckIntervalData currentData, final CheckIntervalData nextData) {
            return nextData.getCheckIntervalStart() != currentData
                    .getCheckIntervalStart()
                    && currentData.getEventCount() < breaker.getClosingThreshold();
        }
    }

    /** A map for accessing the strategy objects for the different states. */
    private static final Map<State, AbstractStateStrategy> STRATEGY_MAP = createStrategyMap();

    /**
     * Creates the map with strategy objects. It allows access for a strategy for a given
     * state.
     *
     * @return The strategy map
     */
    private static Map<State, AbstractStateStrategy> createStrategyMap() {
        final Map<State, AbstractStateStrategy> map = new EnumMap<>(State.class);
        map.put(State.CLOSED, new StateStrategyClosed());
        map.put(State.OPEN, new StateStrategyOpen());
        return map;
    }

    /**
     * Returns the {@link AbstractStateStrategy} object responsible for the given state.
     *
     * @param state The state
     * @return The corresponding {@link AbstractStateStrategy}
     * @throws CircuitBreakingException Thrown if the strategy cannot be resolved.
     */
    private static AbstractStateStrategy stateStrategy(final State state) {
        return STRATEGY_MAP.get(state);
    }

    /** Stores information about the current check interval. */
    private final AtomicReference<CheckIntervalData> checkIntervalData;

    /** The threshold for opening the circuit breaker. */
    private final int openingThreshold;

    /** The time interval for opening the circuit breaker. */
    private final long openingInterval;

    /** The threshold for closing the circuit breaker. */
    private final int closingThreshold;

    /** The time interval for closing the circuit breaker. */
    private final long closingInterval;

    /**
     * Creates a new instance of {@link EventCountCircuitBreaker} which uses the same parameters for
     * opening and closing checks.
     *
     * @param threshold The threshold for changing the status of the circuit breaker; if
     * the number of events received in a check interval is greater than this value, the
     * circuit breaker is opened; if it is lower than this value, it is closed again
     * @param checkInterval The check interval for opening or closing the circuit breaker
     * @param checkUnit The {@link TimeUnit} defining the check interval
     */
    public EventCountCircuitBreaker(final int threshold, final long checkInterval, final TimeUnit checkUnit) {
        this(threshold, checkInterval, checkUnit, threshold);
    }

    /**
     * Creates a new instance of {@link EventCountCircuitBreaker} with the same interval for opening
     * and closing checks.
     *
     * @param openingThreshold The threshold for opening the circuit breaker; if this
     * number of events is received in the time span determined by the check interval, the
     * circuit breaker is opened
     * @param checkInterval The check interval for opening or closing the circuit breaker
     * @param checkUnit The {@link TimeUnit} defining the check interval
     * @param closingThreshold The threshold for closing the circuit breaker; if the
     * number of events received in the time span determined by the check interval goes
     * below this threshold, the circuit breaker is closed again
     */
    public EventCountCircuitBreaker(final int openingThreshold, final long checkInterval, final TimeUnit checkUnit,
                                    final int closingThreshold) {
        this(openingThreshold, checkInterval, checkUnit, closingThreshold, checkInterval,
                checkUnit);
    }

    /**
     * Creates a new instance of {@link EventCountCircuitBreaker} and initializes all properties for
     * opening and closing it based on threshold values for events occurring in specific
     * intervals.
     *
     * @param openingThreshold The threshold for opening the circuit breaker; if this
     * number of events is received in the time span determined by the opening interval,
     * the circuit breaker is opened
     * @param openingInterval The interval for opening the circuit breaker
     * @param openingUnit The {@link TimeUnit} defining the opening interval
     * @param closingThreshold The threshold for closing the circuit breaker; if the
     * number of events received in the time span determined by the closing interval goes
     * below this threshold, the circuit breaker is closed again
     * @param closingInterval The interval for closing the circuit breaker
     * @param closingUnit The {@link TimeUnit} defining the closing interval
     */
    public EventCountCircuitBreaker(final int openingThreshold, final long openingInterval,
                                    final TimeUnit openingUnit, final int closingThreshold, final long closingInterval,
                                    final TimeUnit closingUnit) {
        checkIntervalData = new AtomicReference<>(new CheckIntervalData(0, 0));
        this.openingThreshold = openingThreshold;
        this.openingInterval = openingUnit.toNanos(openingInterval);
        this.closingThreshold = closingThreshold;
        this.closingInterval = closingUnit.toNanos(closingInterval);
    }

    /**
     * Changes the state of this circuit breaker and also initializes a new
     * {@link CheckIntervalData} object.
     *
     * @param newState The new state to be set
     */
    private void changeStateAndStartNewCheckInterval(final State newState) {
        changeState(newState);
        checkIntervalData.set(new CheckIntervalData(0, nanoTime()));
    }

    /**
     * {@inheritDoc}
     * <p>
     * This implementation checks the internal event counter against the
     * threshold values and the check intervals. This may cause a state change of this
     * circuit breaker.
     * </p>
     */
    @Override
    public boolean checkState() {
        return performStateCheck(0);
    }

    /**
     * {@inheritDoc}
     * <p>
     * A new check interval is started. If too many events are received in
     * this interval, the circuit breaker changes again to state open. If this circuit
     * breaker is already closed, this method has no effect, except that a new check
     * interval is started.
     * </p>
     */
    @Override
    public void close() {
        super.close();
        checkIntervalData.set(new CheckIntervalData(0, nanoTime()));
    }

    /**
     * Gets the interval (in nanoseconds) for checking for the closing threshold.
     *
     * @return The opening check interval
     */
    public long getClosingInterval() {
        return closingInterval;
    }

    /**
     * Gets the threshold value for closing the circuit breaker. If the number of
     * events received in the time span determined by the closing interval goes below this
     * threshold, the circuit breaker is closed again.
     *
     * @return The closing threshold
     */
    public int getClosingThreshold() {
        return closingThreshold;
    }

    /**
     * Gets the interval (in nanoseconds) for checking for the opening threshold.
     *
     * @return The opening check interval
     */
    public long getOpeningInterval() {
        return openingInterval;
    }

    /**
     * Gets the threshold value for opening the circuit breaker. If this number of
     * events is received in the time span determined by the opening interval, the circuit
     * breaker is opened.
     *
     * @return The opening threshold
     */
    public int getOpeningThreshold() {
        return openingThreshold;
    }

    /**
     * Increments the monitored value by <strong>1</strong> and performs a check of the current state of this
     * circuit breaker. This method works like {@link #checkState()}, but the monitored
     * value is incremented before the state check is performed.
     *
     * @return <strong>true</strong> if the circuit breaker is now closed;
     * <strong>false</strong> otherwise
     */
    public boolean incrementAndCheckState() {
        return incrementAndCheckState(1);
    }

    /**
     * {@inheritDoc}
     *
     * <p>
     * The event count is a protective counter: negative increments are rejected, and an increment that would overflow
     * {@link Long#MAX_VALUE} saturates the counter at {@link Long#MAX_VALUE} instead of wrapping negative. Saturation prevents
     * an overloaded interval from causing an automatic close. The counter resets when a new check interval starts or the
     * circuit breaker changes state.
     * </p>
     *
     * @throws IllegalArgumentException Thrown if the increment is negative.
     */
    @Override
    public boolean incrementAndCheckState(final Integer increment) {
        if (increment.intValue() < 0) {
            throw new IllegalArgumentException("Increment must not be negative: " + increment);
        }
        return performStateCheck(increment);
    }

    /**
     * Returns the current time in nanoseconds. This method is used to obtain the current
     * time. This is needed to calculate the check intervals correctly.
     *
     * @return The current time in nanoseconds
     */
    long nanoTime() {
        return System.nanoTime();
    }

    /**
     * Calculates the next {@link CheckIntervalData} object based on the current data and
     * the current state. The next data object takes the counter increment and the current
     * time into account.
     *
     * @param increment The increment for the internal counter
     * @param currentData The current check data object
     * @param currentState The current state of the circuit breaker
     * @param time The current time
     * @return The updated {@link CheckIntervalData} object
     */
    private CheckIntervalData nextCheckIntervalData(final int increment,
            final CheckIntervalData currentData, final State currentState, final long time) {
        final CheckIntervalData nextData;
        if (stateStrategy(currentState).isCheckIntervalFinished(this, currentData, time)) {
            nextData = new CheckIntervalData(increment, time);
        } else {
            nextData = currentData.increment(increment);
        }
        return nextData;
    }

    /**
     * {@inheritDoc}
     * <p>
     * This circuit breaker may close itself again if the number of events
     * received during a check interval goes below the closing threshold. If this circuit
     * breaker is already open, this method has no effect, except that a new check
     * interval is started.
     * </p>
     */
    @Override
    public void open() {
        super.open();
        checkIntervalData.set(new CheckIntervalData(0, nanoTime()));
    }

    /**
     * Actually checks the state of this circuit breaker and executes a state transition
     * if necessary.
     *
     * @param increment The increment for the internal counter
     * @return A flag whether the circuit breaker is now closed
     */
    private boolean performStateCheck(final int increment) {
        CheckIntervalData currentData;
        CheckIntervalData nextData;
        State currentState;

        do {
            final long time = nanoTime();
            currentState = state.get();
            currentData = checkIntervalData.get();
            nextData = nextCheckIntervalData(increment, currentData, currentState, time);
        } while (!updateCheckIntervalData(currentData, nextData));

        // This might cause a race condition if other changes happen in between!
        // Refer to the header comment!
        if (stateStrategy(currentState).isStateTransition(this, currentData, nextData)) {
            currentState = currentState.oppositeState();
            changeStateAndStartNewCheckInterval(currentState);
        }
        return !isOpen(currentState);
    }

    /**
     * Updates the {@link CheckIntervalData} object. The current data object is replaced
     * by the one modified by the last check. The return value indicates whether this was
     * successful. If it is <strong>false</strong>, another thread interfered, and the
     * whole operation has to be redone.
     *
     * @param currentData The current check data object
     * @param nextData The replacing check data object
     * @return A flag whether the update was successful
     */
    private boolean updateCheckIntervalData(final CheckIntervalData currentData,
            final CheckIntervalData nextData) {
        return currentData == nextData
                || checkIntervalData.compareAndSet(currentData, nextData);
    }

}