You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@commons.apache.org by pa...@apache.org on 2016/06/07 20:22:49 UTC
[2/2] [lang] LANG-1085: Add a circuit breaker implementation (closes
#95)
LANG-1085: Add a circuit breaker implementation (closes #95)
Squashed commit of the following:
commit 1b12db706b457408562a166c21bc6470a8064620
Merge: edebd97 def7832
Author: Oliver Heger <ol...@oliver-heger.de>
Date: Wed Jun 10 21:27:17 2015 +0200
Merge pull request #1 from kinow/patch1
This commit:
commit def78320c5efcb734572c5b48623a952e3c94c75
Author: Bruno P. Kinoshita <br...@yahoo.com.br>
Date: Wed Jun 10 22:57:33 2015 +1200
This commit:
* fixes JavaDocs warnings in Checkstyle
* adds license headers
* renames the MemoryCircuitBreaker to ThresholdCircuitBreaker
* adds more tests to have a good coverage
commit edebd970582ce2984258352c8482bcb51587f7d2
Author: oheger <ol...@oliver-heger.de>
Date: Tue Jun 9 22:11:11 2015 +0200
Documentation updates for EventCountCircuitBreaker.
commit 3d8a4ff85a78173dd8f5437d9b4d931069128e20
Author: oheger <ol...@oliver-heger.de>
Date: Tue Jun 9 21:55:02 2015 +0200
Renamed TimedCircuitBreaker to EventCountCircuitBreaker.
This name seems to be more appropriate as the main property of this circuit
breaker implementation is that it counts specific events and changes its state
when a certain threshold is reached.
commit 0e4ce5a984ef699ddace61bbf430d695b011c6b5
Author: oheger <ol...@oliver-heger.de>
Date: Tue Jun 9 21:51:22 2015 +0200
Added documentation for CircuitBreaker interface.
commit 0e6b22190ef1868655fec9b3814a2345ad206f76
Author: oheger <ol...@oliver-heger.de>
Date: Mon Jun 8 22:27:05 2015 +0200
Initial import of CircuitBreaker classes.
Project: http://git-wip-us.apache.org/repos/asf/commons-lang/repo
Commit: http://git-wip-us.apache.org/repos/asf/commons-lang/commit/89eea1e7
Tree: http://git-wip-us.apache.org/repos/asf/commons-lang/tree/89eea1e7
Diff: http://git-wip-us.apache.org/repos/asf/commons-lang/diff/89eea1e7
Branch: refs/heads/master
Commit: 89eea1e72b6c9f9faa986fbe548a4fb1914cca9d
Parents: 6c687c7
Author: oheger <ol...@oliver-heger.de>
Authored: Thu Jun 11 21:39:40 2015 +0200
Committer: pascalschumacher <pa...@gmx.net>
Committed: Tue Jun 7 22:22:40 2016 +0200
----------------------------------------------------------------------
.../concurrent/AbstractCircuitBreaker.java | 164 ++++++
.../lang3/concurrent/CircuitBreaker.java | 93 +++
.../concurrent/CircuitBreakingException.java | 66 +++
.../concurrent/EventCountCircuitBreaker.java | 567 +++++++++++++++++++
.../concurrent/ThresholdCircuitBreaker.java | 127 +++++
.../EventCountCircuitBreakerTest.java | 401 +++++++++++++
.../concurrent/ThresholdCircuitBreakerTest.java | 87 +++
7 files changed, 1505 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/commons-lang/blob/89eea1e7/src/main/java/org/apache/commons/lang3/concurrent/AbstractCircuitBreaker.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/commons/lang3/concurrent/AbstractCircuitBreaker.java b/src/main/java/org/apache/commons/lang3/concurrent/AbstractCircuitBreaker.java
new file mode 100644
index 0000000..6df8642
--- /dev/null
+++ b/src/main/java/org/apache/commons/lang3/concurrent/AbstractCircuitBreaker.java
@@ -0,0 +1,164 @@
+/*
+ * 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
+ *
+ * http://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.beans.PropertyChangeSupport;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * Base class for circuit breakers.
+ *
+ * @param <T> the type of the value monitored by this circuit breaker
+ */
+public abstract class AbstractCircuitBreaker<T> implements CircuitBreaker<T> {
+ /**
+ * The name of the <em>open</em> property as it is passed to registered
+ * change listeners.
+ */
+ public static final String PROPERTY_NAME = "open";
+
+ /** The current state of this circuit breaker. */
+ protected final AtomicReference<State> state = new AtomicReference<State>(State.CLOSED);
+
+ /** An object for managing change listeners registered at this instance. */
+ private final PropertyChangeSupport changeSupport;
+
+ /**
+ * Creates an {@AbstractCircuitBreaker}. It also creates an internal {@code PropertyChangeSupport}.
+ */
+ public AbstractCircuitBreaker() {
+ changeSupport = new PropertyChangeSupport(this);
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ public boolean isOpen() {
+ return isOpen(state.get());
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ public boolean isClosed() {
+ return !isOpen();
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ public abstract boolean checkState();
+
+ /**
+ * {@inheritDoc}
+ */
+ public abstract boolean incrementAndCheckState(T increment);
+
+ /**
+ * {@inheritDoc}
+ */
+ public void close() {
+ changeState(State.CLOSED);
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ public void open() {
+ changeState(State.OPEN);
+ }
+
+ /**
+ * Converts the given state value to a boolean <em>open</em> property.
+ *
+ * @param state the state to be converted
+ * @return the boolean open flag
+ */
+ protected static boolean isOpen(State state) {
+ return state == State.OPEN;
+ }
+
+ /**
+ * Changes the internal state of this circuit breaker. If there is actually a change
+ * of the state value, all registered change listeners are notified.
+ *
+ * @param newState the new state to be set
+ */
+ protected void changeState(State newState) {
+ if (state.compareAndSet(newState.oppositeState(), newState)) {
+ changeSupport.firePropertyChange(PROPERTY_NAME, !isOpen(newState), isOpen(newState));
+ }
+ }
+
+ /**
+ * Adds a change listener to this circuit breaker. This listener is notified whenever
+ * the state of this circuit breaker changes. If the listener is
+ * <strong>null</strong>, it is silently ignored.
+ *
+ * @param listener the listener to be added
+ */
+ public void addChangeListener(PropertyChangeListener listener) {
+ changeSupport.addPropertyChangeListener(listener);
+ }
+
+ /**
+ * Removes the specified change listener from this circuit breaker.
+ *
+ * @param listener the listener to be removed
+ */
+ public void removeChangeListener(PropertyChangeListener listener) {
+ changeSupport.removePropertyChangeListener(listener);
+ }
+
+ /**
+ * An internal enumeration representing the different states of a circuit
+ * breaker. This class also contains some logic for performing state
+ * transitions. This is done to avoid complex if-conditions in the code of
+ * {@code CircuitBreaker}.
+ */
+ protected static enum State {
+ CLOSED {
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public State oppositeState() {
+ return OPEN;
+ }
+ },
+
+ OPEN {
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public State oppositeState() {
+ return CLOSED;
+ }
+ };
+
+ /**
+ * Returns the opposite state to the represented state. This is useful
+ * for flipping the current state.
+ *
+ * @return the opposite state
+ */
+ public abstract State oppositeState();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/commons-lang/blob/89eea1e7/src/main/java/org/apache/commons/lang3/concurrent/CircuitBreaker.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/commons/lang3/concurrent/CircuitBreaker.java b/src/main/java/org/apache/commons/lang3/concurrent/CircuitBreaker.java
new file mode 100644
index 0000000..780af46
--- /dev/null
+++ b/src/main/java/org/apache/commons/lang3/concurrent/CircuitBreaker.java
@@ -0,0 +1,93 @@
+/*
+ * 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
+ *
+ * http://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;
+
+/**
+ * <p>
+ * An interface describing a <a
+ * href="http://martinfowler.com/bliki/CircuitBreaker.html">Circuit Breaker</a> component.
+ * </p>
+ * <p>
+ * A <em>circuit breaker</em> can be used to protect an application against unreliable
+ * services or unexpected load. It typically monitors a specific resource. As long as this
+ * resource works as expected, it stays in state <em>closed</em>, meaning that the
+ * resource can be used. If problems are encountered when using the resource, the circuit
+ * breaker can switch into state <em>open</em>; then access to this resource is
+ * prohibited. Depending on a concrete implementation, it is possible that the circuit
+ * breaker switches back to state <em>closed</em> when the resource becomes available
+ * again.
+ * </p>
+ * <p>
+ * This interface defines a generic protocol of a circuit breaker component. It should be
+ * sufficiently generic to be applied to multiple different use cases.
+ * </p>
+ *
+ * @param <T> the type of the value monitored by this circuit breaker
+ */
+public interface CircuitBreaker<T> {
+ /**
+ * Returns the current open state of this circuit breaker. A return value of
+ * <strong>true</strong> means that the circuit breaker is currently open indicating a
+ * problem in the monitored sub system.
+ *
+ * @return the current open state of this circuit breaker
+ */
+ boolean isOpen();
+
+ /**
+ * Returns the current closed state of this circuit breaker. A return value of
+ * <strong>true</strong> means that the circuit breaker is currently closed. This
+ * means that everything is okay with the monitored sub system.
+ *
+ * @return the current closed state of this circuit breaker
+ */
+ boolean isClosed();
+
+ /**
+ * Checks the state of this circuit breaker and changes it if necessary. The return
+ * value indicates whether the circuit breaker is now in state {@code CLOSED}; a value
+ * of <strong>true</strong> typically means that the current operation can continue.
+ *
+ * @return <strong>true</strong> if the circuit breaker is now closed;
+ * <strong>false</strong> otherwise
+ */
+ boolean checkState();
+
+ /**
+ * Closes this circuit breaker. Its state is changed to closed. If this circuit
+ * breaker is already closed, this method has no effect.
+ */
+ void close();
+
+ /**
+ * Opens this circuit breaker. Its state is changed to open. Depending on a concrete
+ * implementation, it may close itself again if the monitored sub system becomes
+ * available. If this circuit breaker is already open, this method has no effect.
+ */
+ void open();
+
+ /**
+ * Increments the monitored value 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.
+ *
+ * @param increment value to increment in the monitored value of the circuit breaker
+ * @return <strong>true</strong> if the circuit breaker is now closed;
+ * <strong>false</strong> otherwise
+ */
+ boolean incrementAndCheckState(T increment);
+}
http://git-wip-us.apache.org/repos/asf/commons-lang/blob/89eea1e7/src/main/java/org/apache/commons/lang3/concurrent/CircuitBreakingException.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/commons/lang3/concurrent/CircuitBreakingException.java b/src/main/java/org/apache/commons/lang3/concurrent/CircuitBreakingException.java
new file mode 100644
index 0000000..e6cbdf7
--- /dev/null
+++ b/src/main/java/org/apache/commons/lang3/concurrent/CircuitBreakingException.java
@@ -0,0 +1,66 @@
+/*
+ * 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
+ *
+ * http://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;
+
+/**
+ * <p>
+ * An exception class used for reporting runtime error conditions related to
+ * circuit breakers.
+ * </p>
+ */
+public class CircuitBreakingException extends RuntimeException {
+ /**
+ * The serial version UID.
+ */
+ private static final long serialVersionUID = 1408176654686913340L;
+
+ /**
+ * Creates a new, uninitialized instance of {@code CircuitBreakingException}.
+ */
+ public CircuitBreakingException() {
+ super();
+ }
+
+ /**
+ * Creates a new instance of {@code CircuitBreakingException} and initializes it with the given message and cause.
+ *
+ * @param message the error message
+ * @param cause the cause of this exception
+ */
+ public CircuitBreakingException(String message, Throwable cause) {
+ super(message, cause);
+ }
+
+ /**
+ * Creates a new instance of {@code CircuitBreakingException} and initializes it with the given message.
+ *
+ * @param message the error message
+ */
+ public CircuitBreakingException(String message) {
+ super(message);
+ }
+
+ /**
+ * Creates a new instance of {@code CircuitBreakingException} and initializes it with the given cause.
+ *
+ * @param cause the cause of this exception
+ */
+ public CircuitBreakingException(Throwable cause) {
+ super(cause);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/commons-lang/blob/89eea1e7/src/main/java/org/apache/commons/lang3/concurrent/EventCountCircuitBreaker.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/commons/lang3/concurrent/EventCountCircuitBreaker.java b/src/main/java/org/apache/commons/lang3/concurrent/EventCountCircuitBreaker.java
new file mode 100644
index 0000000..312c3b4
--- /dev/null
+++ b/src/main/java/org/apache/commons/lang3/concurrent/EventCountCircuitBreaker.java
@@ -0,0 +1,567 @@
+/*
+ * 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
+ *
+ * http://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.util.EnumMap;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * <p>
+ * A simple implementation of the <a
+ * href="http://martinfowler.com/bliki/CircuitBreaker.html">Circuit Breaker</a> pattern
+ * that counts specific events.
+ * </p>
+ * <p>
+ * A <em>circuit breaker</em> can be used to protect an application against unreliable
+ * services or unexpected load. A newly created {@code 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 {@code 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
+ * {@code 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 {@code 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 {@code 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>
+ */
+public class EventCountCircuitBreaker extends AbstractCircuitBreaker<Integer> {
+
+ /** A map for accessing the strategy objects for the different states. */
+ private static final Map<State, StateStrategy> STRATEGY_MAP = createStrategyMap();
+
+ /** 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 {@code 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 {@code 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 {@code TimeUnit} defining the closing interval
+ */
+ public EventCountCircuitBreaker(int openingThreshold, long openingInterval,
+ TimeUnit openingUnit, int closingThreshold, long closingInterval,
+ TimeUnit closingUnit) {
+ super();
+ checkIntervalData = new AtomicReference<CheckIntervalData>(new CheckIntervalData(0, 0));
+ this.openingThreshold = openingThreshold;
+ this.openingInterval = openingUnit.toNanos(openingInterval);
+ this.closingThreshold = closingThreshold;
+ this.closingInterval = closingUnit.toNanos(closingInterval);
+ }
+
+ /**
+ * Creates a new instance of {@code 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 {@code 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(int openingThreshold, long checkInterval, TimeUnit checkUnit,
+ int closingThreshold) {
+ this(openingThreshold, checkInterval, checkUnit, closingThreshold, checkInterval,
+ checkUnit);
+ }
+
+ /**
+ * Creates a new instance of {@code 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 {@code TimeUnit} defining the check interval
+ */
+ public EventCountCircuitBreaker(int threshold, long checkInterval, TimeUnit checkUnit) {
+ this(threshold, checkInterval, checkUnit, threshold);
+ }
+
+ /**
+ * Returns 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;
+ }
+
+ /**
+ * Returns the interval (in nanoseconds) for checking for the opening threshold.
+ *
+ * @return the opening check interval
+ */
+ public long getOpeningInterval() {
+ return openingInterval;
+ }
+
+ /**
+ * Returns 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;
+ }
+
+ /**
+ * Returns the interval (in nanoseconds) for checking for the closing threshold.
+ *
+ * @return the opening check interval
+ */
+ public long getClosingInterval() {
+ return closingInterval;
+ }
+
+ /**
+ * {@inheritDoc} 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.
+ */
+ @Override
+ public boolean checkState() {
+ return performStateCheck(0);
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public boolean incrementAndCheckState(Integer increment)
+ throws CircuitBreakingException {
+ return performStateCheck(1);
+ }
+
+ /**
+ * 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} 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.
+ */
+ @Override
+ public void open() {
+ super.open();
+ checkIntervalData.set(new CheckIntervalData(0, now()));
+ }
+
+ /**
+ * {@inheritDoc} 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.
+ */
+ @Override
+ public void close() {
+ super.close();
+ checkIntervalData.set(new CheckIntervalData(0, now()));
+ }
+
+ /**
+ * 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(int increment) {
+ CheckIntervalData currentData;
+ CheckIntervalData nextData;
+ State currentState;
+
+ do {
+ long time = now();
+ 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 {@code 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(CheckIntervalData currentData,
+ CheckIntervalData nextData) {
+ return currentData == nextData
+ || checkIntervalData.compareAndSet(currentData, nextData);
+ }
+
+ /**
+ * Changes the state of this circuit breaker and also initializes a new
+ * {@code CheckIntervalData} object.
+ *
+ * @param newState the new state to be set
+ */
+ private void changeStateAndStartNewCheckInterval(State newState) {
+ changeState(newState);
+ checkIntervalData.set(new CheckIntervalData(0, now()));
+ }
+
+ /**
+ * Calculates the next {@code 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 {@code CheckIntervalData} object
+ */
+ private CheckIntervalData nextCheckIntervalData(int increment,
+ CheckIntervalData currentData, State currentState, long time) {
+ CheckIntervalData nextData;
+ if (stateStrategy(currentState).isCheckIntervalFinished(this, currentData, time)) {
+ nextData = new CheckIntervalData(increment, time);
+ } else {
+ nextData = currentData.increment(increment);
+ }
+ return nextData;
+ }
+
+ /**
+ * 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 now() {
+ return System.nanoTime();
+ }
+
+ /**
+ * Returns the {@code StateStrategy} object responsible for the given state.
+ *
+ * @param state the state
+ * @return the corresponding {@code StateStrategy}
+ * @throws CircuitBreakingException if the strategy cannot be resolved
+ */
+ private static StateStrategy stateStrategy(State state) {
+ StateStrategy strategy = STRATEGY_MAP.get(state);
+ return strategy;
+ }
+
+ /**
+ * Creates the map with strategy objects. It allows access for a strategy for a given
+ * state.
+ *
+ * @return the strategy map
+ */
+ private static Map<State, StateStrategy> createStrategyMap() {
+ Map<State, StateStrategy> map = new EnumMap<State, StateStrategy>(State.class);
+ map.put(State.CLOSED, new StateStrategyClosed());
+ map.put(State.OPEN, new StateStrategyOpen());
+ return map;
+ }
+
+ /**
+ * 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 class CheckIntervalData {
+ /** The counter for events. */
+ private final int eventCount;
+
+ /** The start time of the current check interval. */
+ private final long checkIntervalStart;
+
+ /**
+ * Creates a new instance of {@code CheckIntervalData}.
+ *
+ * @param count the current count value
+ * @param intervalStart the start time of the check interval
+ */
+ public CheckIntervalData(int count, long intervalStart) {
+ eventCount = count;
+ checkIntervalStart = intervalStart;
+ }
+
+ /**
+ * Returns the event counter.
+ *
+ * @return the number of received events
+ */
+ public int getEventCount() {
+ return eventCount;
+ }
+
+ /**
+ * Returns the start time of the current check interval.
+ *
+ * @return the check interval start time
+ */
+ public long getCheckIntervalStart() {
+ return checkIntervalStart;
+ }
+
+ /**
+ * Returns a new instance of {@code CheckIntervalData} with the event counter
+ * incremented by the given delta. If the delta is 0, this object is returned.
+ *
+ * @param delta the delta
+ * @return the updated instance
+ */
+ public CheckIntervalData increment(int delta) {
+ return (delta != 0) ? new CheckIntervalData(getEventCount() + delta,
+ getCheckIntervalStart()) : this;
+ }
+ }
+
+ /**
+ * 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 StateStrategy {
+ /**
+ * Returns a flag whether the end of the current check interval is reached.
+ *
+ * @param breaker the {@code 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(EventCountCircuitBreaker breaker,
+ CheckIntervalData currentData, long now) {
+ return now - currentData.getCheckIntervalStart() > fetchCheckInterval(breaker);
+ }
+
+ /**
+ * Checks whether the specified {@code 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 {@code CircuitBreaker}
+ * @param currentData the current {@code CheckIntervalData} object
+ * @param nextData the updated {@code CheckIntervalData} object
+ * @return a flag whether a state transition should be performed
+ */
+ public abstract boolean isStateTransition(EventCountCircuitBreaker breaker,
+ CheckIntervalData currentData, CheckIntervalData nextData);
+
+ /**
+ * Obtains the check interval to applied for the represented state from the given
+ * {@code CircuitBreaker}.
+ *
+ * @param breaker the {@code CircuitBreaker}
+ * @return the check interval to be applied
+ */
+ protected abstract long fetchCheckInterval(EventCountCircuitBreaker breaker);
+ }
+
+ /**
+ * A specialized {@code StateStrategy} implementation for the state closed.
+ */
+ private static class StateStrategyClosed extends StateStrategy {
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public boolean isStateTransition(EventCountCircuitBreaker breaker,
+ CheckIntervalData currentData, CheckIntervalData nextData) {
+ return nextData.getEventCount() > breaker.getOpeningThreshold();
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ protected long fetchCheckInterval(EventCountCircuitBreaker breaker) {
+ return breaker.getOpeningInterval();
+ }
+ }
+
+ /**
+ * A specialized {@code StateStrategy} implementation for the state open.
+ */
+ private static class StateStrategyOpen extends StateStrategy {
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public boolean isStateTransition(EventCountCircuitBreaker breaker,
+ CheckIntervalData currentData, CheckIntervalData nextData) {
+ return nextData.getCheckIntervalStart() != currentData
+ .getCheckIntervalStart()
+ && currentData.getEventCount() < breaker.getClosingThreshold();
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ protected long fetchCheckInterval(EventCountCircuitBreaker breaker) {
+ return breaker.getClosingInterval();
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/commons-lang/blob/89eea1e7/src/main/java/org/apache/commons/lang3/concurrent/ThresholdCircuitBreaker.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/commons/lang3/concurrent/ThresholdCircuitBreaker.java b/src/main/java/org/apache/commons/lang3/concurrent/ThresholdCircuitBreaker.java
new file mode 100644
index 0000000..06b7152
--- /dev/null
+++ b/src/main/java/org/apache/commons/lang3/concurrent/ThresholdCircuitBreaker.java
@@ -0,0 +1,127 @@
+/*
+ * 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
+ *
+ * http://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.util.concurrent.atomic.AtomicLong;
+
+/**
+ * <p>
+ * A simple implementation of the <a
+ * href="http://martinfowler.com/bliki/CircuitBreaker.html">Circuit Breaker</a> pattern
+ * that opens if the requested increment amount is greater than a given threshold.
+ * </p>
+ *
+ * <p>
+ * It contains an internal counter that starts in zero, and each call increments the counter by a given amount.
+ * If the threshold is zero, the circuit breaker will be in a permanent <em>open</em> state.
+ * </p>
+ *
+ * <p>
+ * An example of use case could be a memory circuit breaker.
+ * </p>
+ *
+ * <pre>
+ * long threshold = 10L;
+ * ThresholdCircuitBreaker breaker = new ThresholdCircuitBreaker(10L);
+ * ...
+ * public void handleRequest(Request request) {
+ * long memoryUsed = estimateMemoryUsage(request);
+ * if (breaker.incrementAndCheckState(memoryUsed)) {
+ * // actually handle this request
+ * } else {
+ * // do something else, e.g. send an error code
+ * }
+ * }
+ * </pre>
+ *
+ * <p>#Thread safe#</p>
+ */
+public class ThresholdCircuitBreaker extends AbstractCircuitBreaker<Long> {
+ /**
+ * The initial value of the internal counter.
+ */
+ private final static long INITIAL_COUNT = 0L;
+
+ /**
+ * The threshold.
+ */
+ private final long threshold;
+
+ /**
+ * Controls the amount used.
+ */
+ private final AtomicLong used;
+
+ /**
+ * <p>Creates a new instance of {@code ThresholdCircuitBreaker} and initializes the threshold.</p>
+ *
+ * @param threshold the threshold.
+ */
+ public ThresholdCircuitBreaker(long threshold) {
+ super();
+ this.used = new AtomicLong(INITIAL_COUNT);
+ this.threshold = threshold;
+ }
+
+ /**
+ * Gets the threshold.
+ *
+ * @return the threshold
+ */
+ public long getThreshold() {
+ return threshold;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public boolean checkState() throws CircuitBreakingException {
+ return isOpen();
+ }
+
+ /**
+ * {@inheritDoc}
+ *
+ * <p>Resets the internal counter back to its initial value (zero).</p>
+ */
+ @Override
+ public void close() {
+ super.close();
+ this.used.set(INITIAL_COUNT);
+ }
+
+ /**
+ * {@inheritDoc}
+ *
+ * <p>If the threshold is zero, the circuit breaker will be in a permanent <em>open</em> state.</p>
+ */
+ @Override
+ public boolean incrementAndCheckState(Long increment) throws CircuitBreakingException {
+ if (threshold == 0) {
+ open();
+ }
+
+ long used = this.used.addAndGet(increment);
+ if (used > threshold) {
+ open();
+ }
+
+ return checkState();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/commons-lang/blob/89eea1e7/src/test/java/org/apache/commons/lang3/concurrent/EventCountCircuitBreakerTest.java
----------------------------------------------------------------------
diff --git a/src/test/java/org/apache/commons/lang3/concurrent/EventCountCircuitBreakerTest.java b/src/test/java/org/apache/commons/lang3/concurrent/EventCountCircuitBreakerTest.java
new file mode 100644
index 0000000..8b685e9
--- /dev/null
+++ b/src/test/java/org/apache/commons/lang3/concurrent/EventCountCircuitBreakerTest.java
@@ -0,0 +1,401 @@
+/*
+ * 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
+ *
+ * http://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 static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.assertTrue;
+
+import java.beans.PropertyChangeEvent;
+import java.beans.PropertyChangeListener;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.Test;
+
+/**
+ * Test class for {@code EventCountCircuitBreaker}.
+ */
+public class EventCountCircuitBreakerTest {
+ /** Constant for the opening threshold. */
+ private static final int OPENING_THRESHOLD = 10;
+
+ /** Constant for the closing threshold. */
+ private static final int CLOSING_THRESHOLD = 5;
+
+ /** Constant for the factor for converting nanoseconds. */
+ private static final long NANO_FACTOR = 1000L * 1000L * 1000L;
+
+ /**
+ * Tests that time units are correctly taken into account by constructors.
+ */
+ @Test
+ public void testIntervalCalculation() {
+ EventCountCircuitBreaker breaker = new EventCountCircuitBreaker(OPENING_THRESHOLD, 1,
+ TimeUnit.SECONDS, CLOSING_THRESHOLD, 2, TimeUnit.MILLISECONDS);
+ assertEquals("Wrong opening interval", NANO_FACTOR, breaker.getOpeningInterval());
+ assertEquals("Wrong closing interval", 2 * NANO_FACTOR / 1000,
+ breaker.getClosingInterval());
+ }
+
+ /**
+ * Tests that the closing interval is the same as the opening interval if it is not
+ * specified.
+ */
+ @Test
+ public void testDefaultClosingInterval() {
+ EventCountCircuitBreaker breaker = new EventCountCircuitBreaker(OPENING_THRESHOLD, 1,
+ TimeUnit.SECONDS, CLOSING_THRESHOLD);
+ assertEquals("Wrong closing interval", NANO_FACTOR, breaker.getClosingInterval());
+ }
+
+ /**
+ * Tests that the closing threshold is the same as the opening threshold if not
+ * specified otherwise.
+ */
+ @Test
+ public void testDefaultClosingThreshold() {
+ EventCountCircuitBreaker breaker = new EventCountCircuitBreaker(OPENING_THRESHOLD, 1,
+ TimeUnit.SECONDS);
+ assertEquals("Wrong closing interval", NANO_FACTOR, breaker.getClosingInterval());
+ assertEquals("Wrong closing threshold", OPENING_THRESHOLD,
+ breaker.getClosingThreshold());
+ }
+
+ /**
+ * Tests that a circuit breaker is closed after its creation.
+ */
+ @Test
+ public void testInitiallyClosed() {
+ EventCountCircuitBreaker breaker = new EventCountCircuitBreaker(OPENING_THRESHOLD, 1,
+ TimeUnit.SECONDS);
+ assertFalse("Open", breaker.isOpen());
+ assertTrue("Not closed", breaker.isClosed());
+ }
+
+ /**
+ * Tests whether the current time is correctly determined.
+ */
+ @Test
+ public void testNow() {
+ EventCountCircuitBreaker breaker = new EventCountCircuitBreaker(OPENING_THRESHOLD, 1,
+ TimeUnit.SECONDS);
+ long now = breaker.now();
+ long delta = Math.abs(System.nanoTime() - now);
+ assertTrue("Delta to current time too large", delta < 100000);
+ }
+
+ /**
+ * Tests that the circuit breaker stays closed if the number of received events stays
+ * below the threshold.
+ */
+ @Test
+ public void testNotOpeningUnderThreshold() {
+ long startTime = 1000;
+ EventCountCircuitBreakerTestImpl breaker = new EventCountCircuitBreakerTestImpl(OPENING_THRESHOLD, 1,
+ TimeUnit.SECONDS, CLOSING_THRESHOLD, 1, TimeUnit.SECONDS);
+ for (int i = 0; i < OPENING_THRESHOLD - 1; i++) {
+ assertTrue("In open state", breaker.at(startTime).incrementAndCheckState());
+ startTime++;
+ }
+ assertTrue("Not closed", breaker.isClosed());
+ }
+
+ /**
+ * Tests that the circuit breaker stays closed if there are a number of received
+ * events, but not in a single check interval.
+ */
+ @Test
+ public void testNotOpeningCheckIntervalExceeded() {
+ long startTime = 0L;
+ long timeIncrement = 3 * NANO_FACTOR / (2 * OPENING_THRESHOLD);
+ EventCountCircuitBreakerTestImpl breaker = new EventCountCircuitBreakerTestImpl(OPENING_THRESHOLD, 1,
+ TimeUnit.SECONDS, CLOSING_THRESHOLD, 1, TimeUnit.SECONDS);
+ for (int i = 0; i < 5 * OPENING_THRESHOLD; i++) {
+ assertTrue("In open state", breaker.at(startTime).incrementAndCheckState());
+ startTime += timeIncrement;
+ }
+ assertTrue("Not closed", breaker.isClosed());
+ }
+
+ /**
+ * Tests that the circuit breaker opens if all conditions are met.
+ */
+ @Test
+ public void testOpeningWhenThresholdReached() {
+ long startTime = 0;
+ long timeIncrement = NANO_FACTOR / OPENING_THRESHOLD - 1;
+ EventCountCircuitBreakerTestImpl breaker = new EventCountCircuitBreakerTestImpl(OPENING_THRESHOLD, 1,
+ TimeUnit.SECONDS, CLOSING_THRESHOLD, 1, TimeUnit.SECONDS);
+ boolean open = false;
+ for (int i = 0; i < OPENING_THRESHOLD + 1; i++) {
+ open = !breaker.at(startTime).incrementAndCheckState();
+ startTime += timeIncrement;
+ }
+ assertTrue("Not open", open);
+ assertFalse("Closed", breaker.isClosed());
+ }
+
+ /**
+ * Tests that an open circuit breaker does not close itself when the number of events
+ * received is over the threshold.
+ */
+ @Test
+ public void testNotClosingOverThreshold() {
+ EventCountCircuitBreakerTestImpl breaker = new EventCountCircuitBreakerTestImpl(OPENING_THRESHOLD,
+ 10, TimeUnit.SECONDS, CLOSING_THRESHOLD, 1, TimeUnit.SECONDS);
+ long startTime = 0;
+ breaker.open();
+ for (int i = 0; i <= CLOSING_THRESHOLD; i++) {
+ assertFalse("Not open", breaker.at(startTime).incrementAndCheckState());
+ startTime += 1000;
+ }
+ assertFalse("Closed in new interval", breaker.at(startTime + NANO_FACTOR)
+ .incrementAndCheckState());
+ assertTrue("Not open at end", breaker.isOpen());
+ }
+
+ /**
+ * Tests that the circuit breaker closes automatically if the number of events
+ * received goes under the closing threshold.
+ */
+ @Test
+ public void testClosingWhenThresholdReached() {
+ EventCountCircuitBreakerTestImpl breaker = new EventCountCircuitBreakerTestImpl(OPENING_THRESHOLD,
+ 10, TimeUnit.SECONDS, CLOSING_THRESHOLD, 1, TimeUnit.SECONDS);
+ breaker.open();
+ breaker.at(1000).incrementAndCheckState();
+ assertFalse("Already closed", breaker.at(2000).checkState());
+ assertFalse("Closed at interval end", breaker.at(NANO_FACTOR).checkState());
+ assertTrue("Not closed after interval end", breaker.at(NANO_FACTOR + 1)
+ .checkState());
+ assertTrue("Not closed at end", breaker.isClosed());
+ }
+
+ /**
+ * Tests whether an explicit open operation fully initializes the internal check data
+ * object. Otherwise, the circuit breaker may close itself directly afterwards.
+ */
+ @Test
+ public void testOpenStartsNewCheckInterval() {
+ EventCountCircuitBreakerTestImpl breaker = new EventCountCircuitBreakerTestImpl(OPENING_THRESHOLD, 2,
+ TimeUnit.SECONDS, CLOSING_THRESHOLD, 1, TimeUnit.SECONDS);
+ breaker.at(NANO_FACTOR - 1000).open();
+ assertTrue("Not open", breaker.isOpen());
+ assertFalse("Already closed", breaker.at(NANO_FACTOR + 100).checkState());
+ }
+
+ /**
+ * Tests whether a new check interval is started if the circuit breaker has a
+ * transition to open state.
+ */
+ @Test
+ public void testAutomaticOpenStartsNewCheckInterval() {
+ EventCountCircuitBreakerTestImpl breaker = new EventCountCircuitBreakerTestImpl(OPENING_THRESHOLD, 2,
+ TimeUnit.SECONDS, CLOSING_THRESHOLD, 1, TimeUnit.SECONDS);
+ long time = 10 * NANO_FACTOR;
+ for (int i = 0; i <= OPENING_THRESHOLD; i++) {
+ breaker.at(time++).incrementAndCheckState();
+ }
+ assertTrue("Not open", breaker.isOpen());
+ time += NANO_FACTOR - 1000;
+ assertFalse("Already closed", breaker.at(time).incrementAndCheckState());
+ time += 1001;
+ assertTrue("Not closed in time interval", breaker.at(time).checkState());
+ }
+
+ /**
+ * Tests whether the circuit breaker can be closed explicitly.
+ */
+ @Test
+ public void testClose() {
+ EventCountCircuitBreakerTestImpl breaker = new EventCountCircuitBreakerTestImpl(OPENING_THRESHOLD, 2,
+ TimeUnit.SECONDS, CLOSING_THRESHOLD, 1, TimeUnit.SECONDS);
+ long time = 0;
+ for (int i = 0; i <= OPENING_THRESHOLD; i++, time += 1000) {
+ breaker.at(time).incrementAndCheckState();
+ }
+ assertTrue("Not open", breaker.isOpen());
+ breaker.close();
+ assertTrue("Not closed", breaker.isClosed());
+ assertTrue("Open again", breaker.at(time + 1000).incrementAndCheckState());
+ }
+
+ /**
+ * Tests whether events are generated when the state is changed.
+ */
+ @Test
+ public void testChangeEvents() {
+ EventCountCircuitBreaker breaker = new EventCountCircuitBreaker(OPENING_THRESHOLD, 1,
+ TimeUnit.SECONDS);
+ ChangeListener listener = new ChangeListener(breaker);
+ breaker.addChangeListener(listener);
+ breaker.open();
+ breaker.close();
+ listener.verify(Boolean.TRUE, Boolean.FALSE);
+ }
+
+ /**
+ * Tests whether a change listener can be removed.
+ */
+ @Test
+ public void testRemoveChangeListener() {
+ EventCountCircuitBreaker breaker = new EventCountCircuitBreaker(OPENING_THRESHOLD, 1,
+ TimeUnit.SECONDS);
+ ChangeListener listener = new ChangeListener(breaker);
+ breaker.addChangeListener(listener);
+ breaker.open();
+ breaker.removeChangeListener(listener);
+ breaker.close();
+ listener.verify(Boolean.TRUE);
+ }
+
+ /**
+ * Tests that a state transition triggered by multiple threads is handled correctly.
+ * Only the first transition should cause an event to be sent.
+ */
+ @Test
+ public void testStateTransitionGuarded() throws InterruptedException {
+ final EventCountCircuitBreaker breaker = new EventCountCircuitBreaker(OPENING_THRESHOLD, 1,
+ TimeUnit.SECONDS);
+ ChangeListener listener = new ChangeListener(breaker);
+ breaker.addChangeListener(listener);
+
+ final int threadCount = 128;
+ final CountDownLatch latch = new CountDownLatch(1);
+ Thread[] threads = new Thread[threadCount];
+ for (int i = 0; i < threadCount; i++) {
+ threads[i] = new Thread() {
+ @Override
+ public void run() {
+ try {
+ latch.await();
+ } catch (InterruptedException iex) {
+ // ignore
+ }
+ breaker.open();
+ }
+ };
+ threads[i].start();
+ }
+ latch.countDown();
+ for (Thread thread : threads) {
+ thread.join();
+ }
+ listener.verify(Boolean.TRUE);
+ }
+
+ /**
+ * Tests that automatic state transitions generate change events as well.
+ */
+ @Test
+ public void testChangeEventsGeneratedByAutomaticTransitions() {
+ EventCountCircuitBreakerTestImpl breaker = new EventCountCircuitBreakerTestImpl(OPENING_THRESHOLD, 2,
+ TimeUnit.SECONDS, CLOSING_THRESHOLD, 1, TimeUnit.SECONDS);
+ ChangeListener listener = new ChangeListener(breaker);
+ breaker.addChangeListener(listener);
+ long time = 0;
+ for (int i = 0; i <= OPENING_THRESHOLD; i++, time += 1000) {
+ breaker.at(time).incrementAndCheckState();
+ }
+ breaker.at(NANO_FACTOR + 1).checkState();
+ breaker.at(3 * NANO_FACTOR).checkState();
+ listener.verify(Boolean.TRUE, Boolean.FALSE);
+ }
+
+ /**
+ * A test implementation of {@code EventCountCircuitBreaker} which supports mocking the timer.
+ * This is useful for the creation of deterministic tests for switching the circuit
+ * breaker's state.
+ */
+ private static class EventCountCircuitBreakerTestImpl extends EventCountCircuitBreaker {
+ /** The current time in nanoseconds. */
+ private long currentTime;
+
+ public EventCountCircuitBreakerTestImpl(int openingThreshold, long openingInterval,
+ TimeUnit openingUnit, int closingThreshold, long closingInterval,
+ TimeUnit closingUnit) {
+ super(openingThreshold, openingInterval, openingUnit, closingThreshold,
+ closingInterval, closingUnit);
+ }
+
+ /**
+ * Sets the current time to be used by this test object for the next operation.
+ *
+ * @param time the time to set
+ * @return a reference to this object
+ */
+ public EventCountCircuitBreakerTestImpl at(long time) {
+ currentTime = time;
+ return this;
+ }
+
+ /**
+ * {@inheritDoc} This implementation returns the value passed to the {@code at()}
+ * method.
+ */
+ @Override
+ long now() {
+ return currentTime;
+ }
+ }
+
+ /**
+ * A test change listener for checking whether correct change events are generated.
+ */
+ private static class ChangeListener implements PropertyChangeListener {
+ /** The expected event source. */
+ private final Object expectedSource;
+
+ /** A list with the updated values extracted from received change events. */
+ private final List<Boolean> changedValues;
+
+ /**
+ * Creates a new instance of {@code ChangeListener} and sets the expected event
+ * source.
+ *
+ * @param source the expected event source
+ */
+ public ChangeListener(Object source) {
+ expectedSource = source;
+ changedValues = new ArrayList<Boolean>();
+ }
+
+ public void propertyChange(PropertyChangeEvent evt) {
+ assertEquals("Wrong event source", expectedSource, evt.getSource());
+ assertEquals("Wrong property name", "open", evt.getPropertyName());
+ Boolean newValue = (Boolean) evt.getNewValue();
+ Boolean oldValue = (Boolean) evt.getOldValue();
+ assertNotEquals("Old and new value are equal", newValue, oldValue);
+ changedValues.add(newValue);
+ }
+
+ /**
+ * Verifies that change events for the expected values have been received.
+ *
+ * @param values the expected values
+ */
+ public void verify(Boolean... values) {
+ assertArrayEquals(values,
+ changedValues.toArray(new Boolean[changedValues.size()]));
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/commons-lang/blob/89eea1e7/src/test/java/org/apache/commons/lang3/concurrent/ThresholdCircuitBreakerTest.java
----------------------------------------------------------------------
diff --git a/src/test/java/org/apache/commons/lang3/concurrent/ThresholdCircuitBreakerTest.java b/src/test/java/org/apache/commons/lang3/concurrent/ThresholdCircuitBreakerTest.java
new file mode 100644
index 0000000..70f22cd
--- /dev/null
+++ b/src/test/java/org/apache/commons/lang3/concurrent/ThresholdCircuitBreakerTest.java
@@ -0,0 +1,87 @@
+/*
+ * 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
+ *
+ * http://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 static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import org.junit.Test;
+
+/**
+ * Test class for {@code ThresholdCircuitBreaker}.
+ */
+public class ThresholdCircuitBreakerTest {
+
+ /**
+ * Threshold used in tests.
+ */
+ private static final long threshold = 10L;
+
+ private static final long zeroThreshold = 0L;
+
+ /**
+ * Tests that the threshold is working as expected when incremented and no exception is thrown.
+ */
+ @Test
+ public void testThreshold() {
+ ThresholdCircuitBreaker circuit = new ThresholdCircuitBreaker(threshold);
+ circuit.incrementAndCheckState(9L);
+ assertFalse("Circuit opened before reaching the threshold", circuit.incrementAndCheckState(1L));
+ }
+
+ /**
+ * Tests that exceeding the threshold raises an exception.
+ */
+ @Test
+ public void testThresholdCircuitBreakingException() {
+ ThresholdCircuitBreaker circuit = new ThresholdCircuitBreaker(threshold);
+ circuit.incrementAndCheckState(9L);
+ assertTrue("The circuit was spposed to be open after increment above the threshold", circuit.incrementAndCheckState(2L));
+ }
+
+ /**
+ * Test that when threshold is zero, the circuit breaker is always open.
+ */
+ @Test
+ public void testThresholdEqualsZero() {
+ ThresholdCircuitBreaker circuit = new ThresholdCircuitBreaker(zeroThreshold);
+ assertTrue("When the threshold is zero, the circuit is supposed to be always open", circuit.incrementAndCheckState(0L));
+ }
+
+ /**
+ * Tests that closing a {@code ThresholdCircuitBreaker} resets the internal counter.
+ */
+ @Test
+ public void testClosingThresholdCircuitBreaker() {
+ ThresholdCircuitBreaker circuit = new ThresholdCircuitBreaker(threshold);
+ circuit.incrementAndCheckState(9L);
+ circuit.close();
+ // now the internal counter is back at zero, not 9 anymore. So it is safe to increment 9 again
+ assertFalse("Internal counter was not reset back to zero", circuit.incrementAndCheckState(9L));
+ }
+
+ /**
+ * Tests that we can get the threshold value correctly.
+ */
+ @Test
+ public void testGettingThreshold() {
+ ThresholdCircuitBreaker circuit = new ThresholdCircuitBreaker(threshold);
+ assertEquals("Wrong value of threshold", Long.valueOf(threshold), Long.valueOf(circuit.getThreshold()));
+ }
+
+}