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:
- Backoff strategy will be applied only in case of failure
- In case of graph's success It will call my
onCompletemethod withSuccess(_) - In case of
maxRestartsfailures theonCompletewithFailure(_)will be called ?