Feature request: Add `mask`/`onErrorComplete`/`recoverWithComplete`?

[He-Pin]

Just a like .recoverWithRetries(-1, { case _: Throwable => Source.empty }) which recover the error with a complete.

And in fs2 there is a mask method:

def mask: Stream[F, O] = this.handleErrorWith(_ => Stream.empty)

And reactor just add the onErrorComplete in https://github.com/reactor/reactor-core/pull/3159 too.

refs: https://github.com/akka/akka/pull/24985 refs: https://github.com/akka/akka/issues/31455 refs:https://github.com/akka/akka/issues/24951