Как запустить график потоков Akka на отдельном диспетчере с таймаутом?

Question

Этот вопрос основан на проекте любимчика, который я сделал и это SO нить. В определении маршрута HTTP для Akka я запускаю длительный процесс и, естественно, хочу сделать это, не блокируя пользователя. Я могу добиться этого с помощью кода ниже:

blocking-io-dispatcher { 
    type = Dispatcher 
    executor = "thread-pool-executor" 
    thread-pool-executor { 
    fixed-pool-size = 16 
    } 
    throughput = 1 
} 

---

complete { 
    Try(new URL(url)) match { 
    case scala.util.Success(u) => { 
     val src = Source.fromIterator(() => parseMovies(u).iterator) 

     src 
     .via(findMovieByTitleAndYear) 
     .via(persistMovies) 
     .toMat(Sink.fold(Future(0))((acc, elem) => Applicative[Future].map2(acc, elem)(_ + _)))(Keep.right) 
     // run the whole graph on a separate dispatcher 
     .withAttributes(ActorAttributes.dispatcher("blocking-io-dispatcher")) 
     .run.flatten 
     .onComplete { 
      _ match { 
      case scala.util.Success(n) => logger.info(s"Created $n movies") 
      case Failure(t) => logger.error(t, "Failed to process movies") 
      } 
     } 

     Accepted 
    } 
    case Failure(t) => logger.error(t, "Bad URL"); BadRequest -> "Bad URL" 
    } 
} 

чем проблема тогда, если я уже решил это? Проблема в том, что я не уверен, как установить тайм-аут. Выполнение графика создает Future, который выполняется до завершения на выделенном blocking-io-dispatcher. Если я добавлю вызов Await, кодовые блоки. Есть ли способ установить тайм-аут?

Answer 1

completionTimeout этап должен помочь здесь. Пример ниже:

src 
    .completionTimeout(5.seconds) 
    ... 
    .run.flatten 
    .onComplete { 
     case scala.util.Success(n) => logger.info(s"Created $n movies") 
     case Failure(t: TimeoutException) => logger.error(t, "Timed out") 
     case Failure(t) => logger.error(t, "Failed to process movies") 
    } 

Документы, ссылки here.