Random hystrix thoughts

random-hystrix-thoughts.scala

case class HystrixData(group: String, command: String, timeout: Int = 1000)

object Hystrix {
  import com.netflix.hystrix._
  import rx.{Subscription, Subscriber}
  import scala.concurrent.ExecutionContext
  import scala.concurrent.{Promise, Future}

  //B = return value or fallback
  //A = optional value to call the hystrix command with
  type HystCommandNeedsFallback[-A, B] = B => A => Future[B]
  type HystCommand[-A, +B] = A => Future[B]

  def runWithFallback[A, B](data: HystrixData, fallback: => B)(fn: A => B)(implicit executor: ExecutionContext): HystCommand[A, B] =
    run(data)(fn).apply(fallback)

  def runWithoutFallback[A, B](data: HystrixData, fallback: => B)(fn: A => B)(implicit executor: ExecutionContext): HystCommand[A, B] =
    run(data)(fn).apply(throw new UnsupportedOperationException("No fallback available."))

  def run[A, B](data: HystrixData)(fn: A => B)(implicit executor: ExecutionContext): HystCommandNeedsFallback[A, B] =
    (fallback: B) => (param: A) => {
      lazy val fallbackTo = fallback
      lazy val setter = HystrixUtils.commandSetter(data.group, data.command, data.timeout)

      val cmd = new HystrixCommand[B](setter) {
        override def run() =
          fn(param)

        override def getFallback =
          fallback
      }

      val p = Promise[B]()

      //Avoiding conversion to rx.scala.Observable since there's no need to do
      //the implicit conversion.
      val o = cmd.observe()

      val subscription: Subscription = o.subscribe(new Subscriber[B]() {
        override def onNext(result: B): Unit =
          p.success(result)

        override def onError(t: Throwable): Unit =
          p.failure(t)

        override def onCompleted(): Unit =
          ()
      })

      val future = p.future
      future.onComplete(_ => subscription.unsubscribe())
      future
    }
}