Come aggiungere un osservatore basato sul tempo ad un futuro di Scala?

Question

Voglio aggiungere un after(d: FiniteDuration)(callback: => Unit) util a Scala Future s che permetta che io faccia questo:

val f = Future(someTask) 

f.after(30.seconds) { 
    println("f has not completed in 30 seconds!") 
} 

f.after(60.seconds) { 
    println("f has not completed in 60 seconds!") 
} 

Come posso fare questo?

Answer 1

Un modo è quello di utilizzare Future.firstCompletedOf (si veda questo blogpost):

val timeoutFuture = Future { Thread.sleep(500); throw new TimeoutException } 

val f = Future.firstCompletedOf(List(f, timeoutFuture)) 
f.map { case e: TimeoutException => println("f has not completed in 0.5 seconds!") } 

dove TimeoutException è qualche eccezione o tipo.

Answer 2

Utilizzare import akka.pattern.after. Se vuoi implementarlo senza akka, ecco lo source code. L'altro esempio (java) è TimeoutFuture in com.google.common.util.concurrent.

Answer 3

Di solito io uso un pool di thread esecutore e promette:

import scala.concurrent.duration._ 
import java.util.concurrent.{Executors, ScheduledThreadPoolExecutor} 
import scala.concurrent.{Future, Promise} 

val f: Future[Int] = ??? 

val executor = new ScheduledThreadPoolExecutor(2, Executors.defaultThreadFactory(), AbortPolicy) 

def withDelay[T](operation: ⇒ T)(by: FiniteDuration): Future[T] = { 
    val promise = Promise[T]() 
    executor.schedule(new Runnable { 
    override def run() = { 
     promise.complete(Try(operation)) 
    } 
    }, by.length, by.unit) 
    promise.future 
} 

Future.firstCompletedOf(Seq(f, withDelay(println("still going"))(30 seconds))) 
Future.firstCompletedOf(Seq(f, withDelay(println("still still going"))(60 seconds))) 

Answer 4

Qualcosa di simile, forse:

object PimpMyFuture { 
    implicit class PimpedFuture[T](val f: Future[T]) extends AnyVal { 
     def after(delay: FiniteDuration)(callback: => Unit): Future[T] = { 
      Future { 
      blocking { Await.ready(f, delay) } 
      } recover { case _: TimeoutException => callback } 
      f 
     } 
    } 
    } 

    import PimpMyFuture._ 
    Future { Thread.sleep(10000); println ("Done") } 
    .after(5.seconds) { println("Still going") } 

Questa implementazione è semplice, ma raddoppia sostanzialmente il numero di thread necessari - ogni futuro attivo occupa in modo efficace due thread - il che è un po 'dispendioso. In alternativa, è possibile utilizzare le attività pianificate per rendere le vostre attese non bloccanti. Non so di uno scheduler "standard" a Scala (ogni lib ha il proprio), ma per un compito semplice come questo si può usare Java di TimerTask direttamente:

object PimpMyFutureNonBlocking {  
val timer = new java.util.Timer 

implicit class PimpedFuture[T](val f: Future[T]) extends AnyVal { 
    def after(delay: FiniteDuration)(callback: => Unit): Future[T] = { 
     val task = new java.util.TimerTask { 
      def run() { if(!f.isCompleted) callback } 
     } 
     timer.schedule(task, delay.toMillis) 
     f.onComplete { _ => task.cancel } 
     f 
    } 
    } 
}