Add waiting state and clean up tests

src/main/scala/com/hootsuite/circuitbreaker/CircuitBreaker.scala

@@ -14,18 +14,22 @@ import java.util.concurrent.atomic.{AtomicInteger, AtomicReference}
   * A configurable circuit breaker. For the time being, the only state change detection strategy implemented is based on
   * the number of consecutive failures.
   *
-  * @param name the name of the circuit breaker
-  * @param failLimit maximum number of consecutive failures before the circuit breaker is tripped (opened)
-  * @param retryDelay duration until an open/broken circuit breaker lets a call through to verify whether or not it should be reset
-  * @param isResultFailure partial function to allow users to determine return cases which should be considered as failures
+  * @param name                  the name of the circuit breaker
+  * @param failLimit             maximum number of consecutive failures before the circuit breaker is tripped (opened)
+  * @param retryDelay            duration until an open/broken circuit breaker lets a call through to verify whether or not it should be reset
+  * @param isExponentialBackoff  indicating the retry delayed should be increased exponential on consecutive failures
+  * @param exponentialRetryCap   limits the number of times the retryDelay will be increased exponentially, ignored if not exponential backoff
+  * @param isResultFailure       partial function to allow users to determine return cases which should be considered as failures
   * @param isExceptionNotFailure partial function to allow users to determine exceptions which should not be considered failures
-  * @param stateChangeListeners listeners that will be notified when the circuit breaker changes state (open <--> closed)
-  * @param invocationListeners listeners that will be notified whenever the circuit breaker handles a method/function call
+  * @param stateChangeListeners  listeners that will be notified when the circuit breaker changes state (open <--> closed)
+  * @param invocationListeners   listeners that will be notified whenever the circuit breaker handles a method/function call
   */
 class CircuitBreaker private[circuitbreaker] (
   val name: String,
   val failLimit: Int,
   val retryDelay: FiniteDuration,
+  val isExponentialBackoff: Boolean = false,
+  val exponentialRetryCap: Option[Int],
   val isResultFailure: PartialFunction[Any, Boolean] = { case _ => false },
   val isExceptionNotFailure: PartialFunction[Throwable, Boolean] = { case _ => false },
   val stateChangeListeners: List[CircuitBreakerStateChangeListener] = List(),
@@ -38,6 +42,8 @@ class CircuitBreaker private[circuitbreaker] (
       builder.name,
       builder.failLimit,
       builder.retryDelay,
+      builder.isExponentialBackoff,
+      builder.exponentialRetryCap,
       builder.isResultFailure,
       builder.isExceptionNotFailure,
       builder.stateChangeListeners,
@@ -180,9 +186,56 @@ class CircuitBreaker private[circuitbreaker] (
     * @param currentState the expected current state
     * @return true when the state was changed, false when the given state was not the current state
     */
-  def attemptResetBrokenState(currentState: BrokenState): Boolean = {
+  def attemptResetBrokenState(currentState: State, retryCount: Int): Boolean = {
     logger.debug(s"Circuit breaker \'$name\', attempting to reset open/broken state")
-    state.compareAndSet(currentState, new BrokenState(this))
+    val result =
+      state.compareAndSet(currentState, new AttemptResetState(this, retryCount))
+
+    if (result) {
+      stateChangeListeners.foreach { listener =>
+        safely(notificationsExecutionContext)(
+          listener.onAttemptReset,
+          name,
+          "notify listener of reset attempt"
+        )
+      }
+    }
+    result
+  }
+
+  /**
+    * calculate the new retry
+    *
+    * @param retryCount what retry is it
+    * @return the number of ms to wait before retrying
+    */
+  def calcRetryDelay(retryCount: Int): Long = {
+
+    //calc jitter up to 1/10 the current retryDelay
+    //for exponential backoff
+    val jitter: Long = if (this.isExponentialBackoff) {
+      (scala.util.Random.nextFloat() * this.retryDelay.toMillis / 10).toLong
+    } else {
+      0
+    }
+
+    val retryCap = this.exponentialRetryCap match {
+      case Some(x) => x
+      case None => Int.MaxValue
+    }
+
+    val exponent: Int = if (this.isExponentialBackoff) {
+      Math.min(retryCount, retryCap)
+    } else 0
+
+    val result = (this.retryDelay.toMillis + jitter) * Math.pow(2, exponent).toLong
+
+    logger.debug(
+      s"CB retry delay details: jitter $jitter, " +
+        s"retryCap $retryCap, exponent $exponent delay $result"
+    )
+
+    result
   }
 
   /**
@@ -196,16 +249,18 @@ class CircuitBreaker private[circuitbreaker] (
   /**
     * @inheritdoc
     */
-  override def isBroken: Boolean = state.get() match {
-    case _: BrokenState => true
-    case _ => false
-  }
+  override def isBroken: Boolean =
+    state.get() match {
+      case _: BrokenState => true
+      case _: AttemptResetState => true
+      case _ => false
+    }
 
   /**
     * @inheritdoc
     */
   override def isWaiting: Boolean = state.get() match {
-    case s: BrokenState if System.currentTimeMillis() > s.retryAt => true
+    case _: AttemptResetState => true
     case _ => false
   }
 }
@@ -236,7 +291,7 @@ private object CircuitBreaker {
   private def safely(
     ec: ExecutionContext
   )(op: => (String) => Any, str: String, opName: String = "<no operation name specified>"): Unit = {
-    implicit val implicitEc = ec
+    implicit val implicitEc: ExecutionContext = ec
     Future(op(str)).recover {
       case NonFatal(e) =>
         logger.warn(
@@ -286,7 +341,7 @@ private object CircuitBreaker {
     override def onFailure(): Unit =
       incrementFailure()
 
-    private[this] def incrementFailure() = {
+    private[this] def incrementFailure(): Unit = {
       val currentCount = failureCount.incrementAndGet
       logger.debug(
         s"Circuit breaker ${cb.name} increment failure count to $currentCount; fail limit is ${cb.failLimit}"
@@ -299,7 +354,15 @@ private object CircuitBreaker {
     * CircuitBreaker is opened/broken. Invocations fail immediately.
     */
   class BrokenState(cb: CircuitBreaker) extends State {
-    val retryAt: Long = System.currentTimeMillis() + cb.retryDelay.toMillis
+    val retryDelay: Long = cb.calcRetryDelay(0)
+
+    //Automatically transition this state at the retry time
+    implicit val ec: ExecutionContext = ExecutionContext.Implicits.global
+    Future {
+      Thread.sleep(retryDelay)
+    }.onComplete(_ => {
+      cb.attemptResetBrokenState(this, 0)
+    })
 
     override def preInvoke(): Unit = {
       cb.invocationListeners.foreach { listener =>
@@ -310,8 +373,40 @@ private object CircuitBreaker {
         )
       }
 
+      //immediately fail
+      throw new CircuitBreakerBrokenException(
+        cb.name,
+        s"Making ${cb.name} unavailable after ${cb.failLimit} errors"
+      )
+    }
+
+    override def postInvoke(): Unit = { /* do nothing */ }
+
+    override def onThrowable(e: Throwable): Unit = { /* do nothing */ }
+
+    override def onFailure(): Unit = { /* do nothing */ }
+  }
+
+  /**
+    * CircuitBreaker is opened/waiting. Invocations are attempted
+    */
+  class AttemptResetState(cb: CircuitBreaker, retryCount: Int = 0) extends State {
+    val retryAt: Long = retryCount match {
+      case 0 => System.currentTimeMillis()
+      case _ => System.currentTimeMillis() + cb.calcRetryDelay(retryCount)
+    }
+
+    override def preInvoke(): Unit = {
+      cb.invocationListeners.foreach { listener =>
+        safely(cb.notificationsExecutionContext)(
+          listener.onInvocationInAttemptResetState,
+          cb.name,
+          "notify listener of invocation in attempt reset state"
+        )
+      }
+
       val retry = System.currentTimeMillis > retryAt
-      if (!(retry && cb.attemptResetBrokenState(this))) {
+      if (!(retry && cb.attemptResetBrokenState(this, this.retryCount + 1))) {
         throw new CircuitBreakerBrokenException(
           cb.name,
           s"Making ${cb.name} unavailable after ${cb.failLimit} errors"
@@ -328,23 +423,28 @@ private object CircuitBreaker {
 
     override def onFailure(): Unit = { /* do nothing */ }
   }
+
 }
 
 /**
   * Builder for [[CircuitBreaker]]
   *
-  * @param name the name of the circuit breaker
-  * @param failLimit maximum number of consecutive failures before the circuit breaker is tripped (opened)
-  * @param retryDelay duration until an open/broken circuit breaker lets a call through to verify whether or not it should be reset
-  * @param isResultFailure partial function to allow users to determine return cases which should be considered as failures
+  * @param name                  the name of the circuit breaker
+  * @param failLimit             maximum number of consecutive failures before the circuit breaker is tripped (opened)
+  * @param retryDelay            duration until an open/broken circuit breaker lets a call through to verify whether or not it should be reset
+  * @param isExponentialBackoff  indicating the retry delayed should be increased exponential on consecutive failures
+  * @param exponentialRetryCap   limits the number of times the retryDelay will be increased exponentially, ignored if not exponential backoff
+  * @param isResultFailure       partial function to allow users to determine return cases which should be considered as failures
   * @param isExceptionNotFailure partial function to allow users to determine exceptions which should not be considered failures
-  * @param stateChangeListeners listeners that will be notified when the circuit breaker changes state (open <--> closed)
-  * @param invocationListeners listeners that will be notified whenever the circuit breaker handles a method/function call
+  * @param stateChangeListeners  listeners that will be notified when the circuit breaker changes state (open <--> closed)
+  * @param invocationListeners   listeners that will be notified whenever the circuit breaker handles a method/function call
   */
 case class CircuitBreakerBuilder(
   name: String,
   failLimit: Int,
   retryDelay: FiniteDuration,
+  isExponentialBackoff: Boolean = false,
+  exponentialRetryCap: Option[Int] = Some(10),
   isResultFailure: PartialFunction[Any, Boolean] = { case _ => false },
   isExceptionNotFailure: PartialFunction[Throwable, Boolean] = { case _ => false },
   stateChangeListeners: List[CircuitBreakerStateChangeListener] = List(),

src/main/scala/com/hootsuite/circuitbreaker/CircuitBreakerRegistry.scala

@@ -12,7 +12,8 @@ object CircuitBreakerRegistry {
 
   private val logger = LoggerFactory.getLogger(getClass)
 
-  private val circuitBreakerStore = emptyCMap[String, ReadOnlyCircuitBreakerSnapshot]
+  private val circuitBreakerStore =
+    emptyCMap[String, ReadOnlyCircuitBreakerSnapshot]
 
   /**
     * Registers a circuit breaker
@@ -43,13 +44,15 @@ object CircuitBreakerRegistry {
     * @param name name of the circuit breaker to remove
     * @return an option value containing the CircuitBreaker, or None if CircuitBreaker was not present in the map before
     */
-  def remove(name: String): Option[ReadOnlyCircuitBreakerSnapshot] = circuitBreakerStore.remove(name)
+  def remove(name: String): Option[ReadOnlyCircuitBreakerSnapshot] =
+    circuitBreakerStore.remove(name)
 
   /**
     * Gets a read-only snapshot of all CircuitBreakers stored in this registry
     * @return a Map containing a read-only snapshot of all CircuitBreakers stored in this registry
     */
-  def getAll: collection.Map[String, ReadOnlyCircuitBreakerSnapshot] = circuitBreakerStore.readOnlySnapshot()
+  def getAll: collection.Map[String, ReadOnlyCircuitBreakerSnapshot] =
+    circuitBreakerStore.readOnlySnapshot()
 
   /**
     * Gets a read-only snapshot of a CircuitBreaker stored in this registry, by its name

src/main/scala/com/hootsuite/circuitbreaker/listeners/CircuitBreakerInvocationListener.scala

@@ -18,4 +18,11 @@ trait CircuitBreakerInvocationListener {
     * @param name - name of the circuit breaker
     */
   def onInvocationInBrokenState(name: String): Unit = { /* empty */ }
+
+  /**
+    * Called when a wrapped function/method is called while the circuit breaker is opened (attempt reset)
+    *
+    * @param name - name of the circuit breaker
+    */
+  def onInvocationInAttemptResetState(name: String): Unit = { /* empty */ }
 }

src/main/scala/com/hootsuite/circuitbreaker/listeners/CircuitBreakerStateChangeListener.scala

@@ -19,6 +19,16 @@ trait CircuitBreakerStateChangeListener {
     */
   def onTrip(name: String): Unit = { /* blank default implementation */ }
 
+  /**
+    * Called when a the circuit breaker is attempting to reset (i.e. open -> waiting)
+    *
+    * @param name - name of the circuit breaker
+    */
+  def onAttemptReset(name: String): Unit = {
+
+    /* blank default implementation */
+  }
+
   /**
     * Called when a the circuit breaker is closed (i.e. open -> closed)
     *

src/test/scala/com/hootsuite/circuitbreaker/CircuitBreakerRegistryTest.scala

@@ -14,7 +14,8 @@ class CircuitBreakerRegistryTest extends FlatSpec with Matchers with BeforeAndAf
 
   val retryDelay = Duration(100, TimeUnit.MILLISECONDS)
 
-  private def waitUntilRetryDelayHasExpired() = Thread.sleep(2 * retryDelay.toMillis)
+  private def waitUntilRetryDelayHasExpired() =
+    Thread.sleep(2 * retryDelay.toMillis)
 
   "registry" should "be empty on startup" in {
     CircuitBreakerRegistry.getAll.isEmpty shouldEqual true
@@ -69,7 +70,8 @@ class CircuitBreakerRegistryTest extends FlatSpec with Matchers with BeforeAndAf
 
   it should "return a read-once version of the underlying circuit breaker" in {
     val name = "trip fast"
-    val actualCircuitBreaker = CircuitBreakerBuilder(name, 1, retryDelay).build()
+    val actualCircuitBreaker =
+      CircuitBreakerBuilder(name, 1, retryDelay).build()
 
     val lookedUpCircuitBreaker =
       CircuitBreakerRegistry.get(name).getOrElse(throw new Exception("should've found this!"))