Jak dodać czasowego obserwatora do Scala Future?

Question

Chcę dodać after(d: FiniteDuration)(callback: => Unit) util do Scala Future s, który pozwoliłby mi na to:

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!") 
} 

Jak mogę to zrobić?

Answer 1

Jednym ze sposobów jest użycie Future.firstCompletedOf (zobacz 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!") } 

gdzie TimeoutException jest jakiś wyjątek lub typ.

Answer 2

Użyj import akka.pattern.after. Jeśli chcesz go zaimplementować bez akka, tutaj jest source code. Innym (java) przykładem jest TimeoutFuture w com.google.common.util.concurrent.

Answer 3

Zwykle używam nici basen wykonawcę i obiecuje:

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

coś takiego, być może:

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") } 

Ta implementacja jest prosta, ale to w zasadzie podwaja liczbę wątków trzeba - każda aktywna przyszłość skutecznie zajmuje dwa wątki - co jest nieco marnotrawstwem. Alternatywnie możesz użyć zaplanowanych zadań, aby twoje oczekiwania nie były blokowane. Nie wiem, od „standardowego” scheduler w Scala (lib każdy ma własne), ale dla prostego zadania jak to można użyć Java TimerTask bezpośrednio:

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 
    } 
    } 
}