【发布时间】:2022-01-19 02:15:53
【问题描述】:
我正在使用 scala 期货来轮询 API 状态。继续轮询直到返回成功或失败。但是,当任何批处理失败时,它应该抛出错误并停止程序。
我无法通过 throw (exception) 或 Future.successful(false) 或 Future.failed(new Exception("error")) 抛出错误。
package FutureScheduler
import scala.concurrent.{Await, Future}
import scala.concurrent.ExecutionContext.Implicits.global
object delayTest extends App {
import scala.concurrent.Future
import scala.concurrent.duration._
import scala.concurrent.ExecutionContext.Implicits.global
def getStatus(batchId: String): String = {
print(s" $batchId ${System.currentTimeMillis()} continue \n")
batchId match {
case "batch1" => "success"
case "batch2" => "failed"
case "batch2" => "success"
}
}
def waitTask(batch: (String, Int)
): Future[Boolean] =
Delayed(x.seconds)(getStatus(batch._1)).flatMap {
case "success" =>
print(s"\n${batch._1} succeeded for ${batch._2}")
Future.successful(true)
case "failed" =>
print(s"\n${batch._1} failed for ${batch._2}")
Future.failed(new Exception("error"))
throw new RuntimeException("errored")
case _ => {
waitTask(batch)
}
}
val statusList = List(Some("batch1", 123), None, Some("batch2", 124)).flatten
val y = 1
val x = 5
try {
Await.ready(Future.traverse(statusList)((waitTask _)), y.minutes)
}
catch {
case e: Exception => println("caught error")
}
print("\nbye now")
}
import scala.concurrent.duration._
import scala.concurrent.{Future, Promise}
object Delayed {
import java.util.{Timer, TimerTask}
private val timer = new Timer
def apply[T](delay: Duration)(task: => T): Future[T] = {
val promise = Promise[T]()
val tt = new TimerTask {
override def run(): Unit = promise.success(task)
}
timer.schedule(tt, delay.toMillis)
promise.future
}
}
【问题讨论】:
-
Delayed定义在哪里?
标签: scala