akka streams RestartFlow playground

RestartFlowPlayground.scala

import akka.actor.ActorSystem
import akka.stream.ActorAttributes
import akka.stream.Supervision.Stop
import akka.stream.scaladsl.{ Flow, RestartFlow, Sink, Source }

import scala.concurrent.Await
import scala.concurrent.duration._
import scala.util.control.NonFatal

import scala.concurrent.Future
import scala.util.Random

object MyApp extends App {

  implicit val as: ActorSystem = ActorSystem()
  val parallelism              = 10

  Await.result(
    Source
      .fromIterator(() => (1 to 10).iterator)
      .via(RestartFlow.onFailuresWithBackoff(1.second, 2.seconds, 0.2, -1)(() => {
        Flow[Int]
          .mapAsync(parallelism)(el => {
            if (Random.nextBoolean()) {
              Future.successful(el)
            } else { Future.failed(throw new IllegalArgumentException("WOOOO")) }
          })
          .addAttributes(ActorAttributes.supervisionStrategy {
            case NonFatal(e) => {
              println(s"LOST ${e}")
              Stop
            }
          })
      }))
      .runWith(Sink.foreach(el => {
        println(s"FINISHED $el")
      })),
    10.seconds
  )

}