Implement Circuit Breaker for Task

[alexandru]

We need a circuit breaker for Task and Future that in case request failures happen, it starts completing subsequent requests with an error for a limited time, in order to give the connection time to recover.

• Akka implementation: http://doc.akka.io/docs/akka/current/common/circuitbreaker.html
• Requested by: https://twitter.com/etorreborre/status/826365202497171457

Implementation from PR #306:

The TaskCircuitBreaker is used to provide stability and prevent cascading failures in distributed systems. This implementation is inspired by the availability of Akka's Circuit Breaker, but note that the implementation and API are different.

Usage:

import monix.eval._
import scala.concurrent.duration._

val circuitBreaker = TaskCircuitBreaker(
  maxFailures = 5,
  resetTimeout = 10.seconds
)

//...
val problematic = Task {
  val nr = util.Random.nextInt()
  if (nr % 2 == 0) nr else
    throw new RuntimeException("dummy")
}

val task = circuitBreaker.protect(problematic.timeout(1.second))

So basically after the maxFailures threshold is reached in a row, then the circuit breaker ends up in the Open state where it keeps fast-failing tasks (with ExecutionRejectedException). It does so until the resetTimeout is reached, when it makes an attempt to reconnect by allowing one task to execute and analyzing its result.

Note in the example that the timeout protection needs to be added per executed Task, since this is a feature of Task already. This is in contrast with Akka's implementation, because that implementation doesn't have the luxury of working with Task 😜

On attempts to reconnect we can also apply an exponential backoff, with a timeout that keeps increasing up to a configured maximum:

val circuitBreaker = TaskCircuitBreaker(
  maxFailures = 5,
  resetTimeout = 10.seconds,
  exponentialBackoffFactor = 2,
  maxResetTimeout = 10.minutes
)