Compose Futures with Recovery in Scala

Question

I have a method, that performs some async actions. Every async action need a recovery. How can I write the method in more composable fashion?

Pseudo scala code:

object Test {

  import ExecutionContext.Implicits.global

  case class ObjectData(
                         internal: String,
                         external1: String = "",
                         external2: String = ""
                       )

  case class ObjectInfo(
                         id: String,
                         internal: String,
                         external1: String = "",
                         external2: String = ""
                       )

  def addObject(data: ObjectData): Future[ObjectInfo] = {
    internalActionWithRollback(data.internal) { objectInfo =>
      externalActionWithRollback(objectInfo.id, data.external1) {
        externalActionWithRollback(objectInfo.id, data.external2) {
          Future.successful(
            objectInfo.copy(
              external1 = data.external1,
              external2 = data.external2
            )
          )
        }
      }
    }
  }

  private def internalActionWithRollback[R](internal: String)
                                           (nextAction: ObjectInfo => Future[R]): Future[R] = ???

  private def externalActionWithRollback[R](id: String, external: String)
                                           (nextAction: => Future[R]): Future[R] = ???

}

Edit:

@IvanKurchenko helped me to solve the problem. We need future wrapper with flatMap and map functions:

object TransactFuture {
  implicit class FutureOps[T, R](underling: Future[T]) {
    def rollbackWith(rollback: PartialFunction[Throwable, Future[R]]): TransactFuture[T, R] = {
      new TransactFuture[T, R](underling, rollback)
    }
    def empty: TransactFuture[T, R] = {
      new TransactFuture[T, R](underling, PartialFunction.empty[Throwable, Future[R]])
    }
  }

}

class TransactFuture[T, R](underlying: Future[T], rollback: PartialFunction[Throwable, Future[R]]) {

  private def recoveryInternal[S](implicit ec: ExecutionContext): PartialFunction[Throwable, Future[S]] = {
    case ex: Throwable =>
      val failed = Future.failed[S](ex)
      rollback.lift(ex).fold(failed)(_.flatMap[S](_ => failed))
  }

  def flatMap[S](f: T => Future[S])(implicit ec: ExecutionContext): Future[S] = {
    underlying.flatMap(f).recoverWith(recoveryInternal)
  }
  def map[S](f: T => S)(implicit ec: ExecutionContext): Future[S] = {
    underlying.map(f).recoverWith(recoveryInternal)
  }
}

With this wrapper we can write the main function:

  def addObject(data: ObjectData): Future[ObjectInfo] = {
    for {
      objectInfo <- addObjectInternal(internal)
      _ <- addExternal(objectInfo.id, data.external1)
        .rollbackWith({
          case _: Throwable => deleteObjectInternal(objectInfo.id)
        })
      _ <- addExternal(objectInfo.id, data.external2)
        .rollbackWith({
          case _: Throwable => deleteExternal(objectInfo.id)
        })
    } yield {
      objectInfo.copy(
        external1 = data.external1,
        external2 = data.external2
      )
    }
  }

user_3298917 · Answer

If I understood you correctly, you want to invoke some action, if child or next Future execution failed. It seems this behavior similar to some kind of transaction. Well, Future out of the box does not provide such behavior, but some wrapper can be implemented instead:

class TransactFuture[T](underlying: Future[T], rollback: PartialFunction[Throwable, Future[Unit]]) {
  def flatMap[S](f: T => Future[S])(implicit executor: ExecutionContext): Future[S] = {
    underlying.flatMap(f).recoverWith {
      case exception: Throwable =>
        val failure = Future.failed[S](exception)
        rollback.lift(exception).fold(failure)(_.flatMap(_ => failure))
    }
  }
}

// Just provides syntax sugar over Future
implicit class FutureOps[T](underling: Future[T]) {
  def rollbackWith(rollback: PartialFunction[Throwable, Future[Unit]]): TransactFuture[T] = {
    new TransactFuture[T](underling, rollback)
  }
}

Which can be used in the next way:

import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.duration._

val action = for {
  _ <- Future.successful(println("A executed")).rollbackWith {
    case _: Throwable => Future.successful(println("A recovered"))
  }
  _ <- Future.successful(println("B executed")).rollbackWith {
    case _: Throwable => Future.successful(println("B recovered"))
  }
  _ <- Future.failed(new Exception("C failed"))
} yield ()

Await.result(action, 1 second)

So the result output will be:

A executed
B executed
B recovered
A recovered
Exception in thread "main" ......
Caused by: java.lang.Exception: C failed

Hope this helps!

Compose Futures with Recovery in Scala

1 Answers