LANG-1085: Add a circuit breaker implementation (closes #95)
Squashed commit of the following:
commit 1b12db706b457408562a166c21bc6470a8064620
Merge: edebd97 def7832
Author: Oliver Heger <oliver.he...@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 <brunodepau...@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 <oliver.he...@oliver-heger.de>
Date: Tue Jun 9 22:11:11 2015 +0200
Documentation updates for EventCountCircuitBreaker.
commit 3d8a4ff85a78173dd8f5437d9b4d931069128e20
Author: oheger <oliver.he...@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 <oliver.he...@oliver-heger.de>
Date: Tue Jun 9 21:51:22 2015 +0200
Added documentation for CircuitBreaker interface.
commit 0e6b22190ef1868655fec9b3814a2345ad206f76
Author: oheger <oliver.he...@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 <oliver.he...@oliver-heger.de>
Authored: Thu Jun 11 21:39:40 2015 +0200
Committer: pascalschumacher <pascalschumac...@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()));
+ }
+
+}
