Comment annuler Future en Scala ?

Question

Java Future a la méthode cancel, qui peut interrompre le thread exécutant la tâche Future. Par exemple, si j'emballe un appel bloquant interruptible dans un Java Future, je peux l'interrompre plus tard.

Scala Future ne fournit pas de méthode cancel. Supposons que j'emballe un appel bloquant interruptible dans un Scala Future. Comment puis-je l'interrompre ?

Answer 1

Ceci n'est pas encore une partie de l'API des Future, mais pourrait être ajouté comme une extension à l'avenir.

En attendant, vous pourriez utiliser le firstCompletedOf pour envelopper 2 futures - le future que vous souhaitez annuler et un future provenant d'une Promise personnalisée. Vous pourriez ensuite annuler le future ainsi créé en échouant la promesse :

def cancellable[T](f: Future[T])(customCode: => Unit): (() => Unit, Future[T]) = {
  val p = Promise[T]
  val first = Future firstCompletedOf Seq(p.future, f)
  val cancellation: () => Unit = {
    () =>
      first onFailure { case e => customCode}
      p failure new Exception
  }
  (cancellation, first)
}

Maintenant, vous pouvez appeler ceci sur n'importe quel future pour obtenir un "wrapper annulable". Exemple d'utilisation :

val f = callReturningAFuture()
val (cancel, f1) = cancellable(f) {
  cancelTheCallReturningAFuture()
}

// quelque part ailleurs dans le code
if (condition) cancel() else println(Await.result(f1))

EDIT :

Pour une discussion détaillée sur l'annulation, consultez le Chapitre 4 dans le livre Learning concurrent programming in Scala.

Answer 2

Je n'ai pas testé cela, mais cela développe la réponse de Pablo Francisco Pérez Hidalgo. Au lieu de bloquer en attendant le java Future, nous utilisons plutôt une Promise intermédiaire.

import java.util.concurrent.{Callable, FutureTask}
import scala.concurrent.{ExecutionContext, Promise}
import scala.util.Try

class Cancellable[T](executionContext: ExecutionContext, todo: => T) {
  private val promise = Promise[T]()

  def future = promise.future

  private val jf: FutureTask[T] = new FutureTask[T](
    new Callable[T] {
      override def call(): T = todo
    }
  ) {
    override def done() = promise.complete(Try(get()))
  }

  def cancel(): Unit = jf.cancel(true)

  executionContext.execute(jf)
}

object Cancellable {
  def apply[T](todo: => T)(implicit executionContext: ExecutionContext): Cancellable[T] =
    new Cancellable[T](executionContext, todo)
}

Answer 3

En annulant, je suppose que vous voudriez interrompre violemment le future.

J'ai trouvé ce segment de code : https://gist.github.com/viktorklang/5409467

J'ai fait quelques tests et cela semble bien fonctionner !

Profitez-en :)

Answer 4

Je pense qu'il est possible de réduire la complexité des implémentations en faisant usage de l'interface Future de Java 7 et de ses implémentations.

Cancellable peut construire un futur Java qui peut être annulé par sa méthode cancel. Un autre futur peut attendre sa complétion devenant ainsi l'interface observable qui est en elle-même immuable dans son état:

 class Cancellable[T](executionContext: ExecutionContext, todo: => T) {

   private val jf: FutureTask[T] = new FutureTask[T](
     new Callable[T] {
       override def call(): T = todo
     }
   )

   executionContext.execute(jf)

   implicit val _: ExecutionContext = executionContext

   val future: Future[T] = Future {
     jf.get
   }

   def cancel(): Unit = jf.cancel(true)

 }

 object Cancellable {
   def apply[T](todo: => T)(implicit executionContext: ExecutionContext): Cancellable[T] =
     new Cancellable[T](executionContext, todo)
 }