Akka streams: how to use backoff strategy only in case of failure?

Viewed 192

I have a rather complex Source source, which I've wrapped into Backoff to be able to restart the stream in case of failure:

val backoffSource = RestartSource.withBackoff(settings){() => source}

I link this source to the sink, which has a materialized value Future[Done]:

val future = backoffSource.toMat(sink)(Keep.right).run()

After that in case of success of this graph I want do some actions when the graph is finished. So, I want to do somethink like:

future.onComplete{
  case Failure(ex) => println(ex)
  case Success(_) => println("Success")
}

But as per documentation RestartSource is

starting an operator again when it fails or completes, each time with a growing time delay between restarts

So my graph will keep restarting until maxRestarts is reached.

How can I implement the next behaviour:

  1. Backoff strategy will be applied only in case of failure
  2. In case of graph's success It will call my onComplete method with Success(_)
  3. In case of maxRestarts failures the onComplete with Failure(_) will be called ?
1 Answers

Use RestartSource.onFailuresWithBackoff:

Wrap the given Source with a Source that will restart it when it fails using an exponential backoff.

From inspection of the source, if the wrapped Source fails enough to fail overall, the stream will fail with the exception from the last failure.

Related