More flow operators

[adamw]

Inspired by https://github.com/apache/pekko/blob/main/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Flow.scala:

• sample(n: Int) -> emit every nth element from the parent flow
• zipWithIndex
• debounce (remove subsequent, repeating elements)
• debounceBy(T => U)
• recover(Throwable => T) (emit one element when the stream fails)
• recoverWith(Throwable => Flow) (start alternative flow on failure)
• recoverWithRetry(RetryConfig)(Throwable => Flow)
• mapWithResource(=> R, R => Option[U])((R, T) => U)
• mapWithCloseableResource(=> R)((R, T) => U)
• scan(V)((V, T) => T)
• conflate
• batch
• expand
• extrapolate
• onErrorComplete
• onErrorRecover

[corem]

@adamw Isn't thebatch behavior already implemented by: def grouped(n: Int): Flow[Seq[T]] ?

ox/core/src/main/scala/ox/flow/FlowOps.scala

Line 614 in 47504fb

def grouped(n: Int): Flow[Seq[T]] = groupedWeighted(n)(_ => 1)

[adamw]

@corem hm batch might be a more challenging one. grouped always groups elements into groups of n elements (except maybe for stream completion), while batch does so only if the downstream consumer is slower. So in normal conditions you might not see any groups at all.

Implementing this would probably require some channels + select.

[He-Pin]

I think we should add onErrorComplete and onErrorResume for Java api, just as rxjava and reactor-core.

@adamw apache/pekko#2116

[adamw]

@He-Pin adding to the list :) If you or anybody else would like to attempt a PR, that would be most welcome!

[halotukozak]

I'd like to to see:

• def partition(predicate: A => Boolean): (Flow[A], Flow[A])
• def unzip[K, V](using A => (K, V)): (Flow[K], Flow[V])
• def unzip3[K, V, W](using A => (K, V, W)): (Flow[K], Flow[V], Flow[W])

also some more runs (not only to List) would be nice, like

• def runToSet(): Set[A]
• def runToMap[K, V]()(using A <:< (K, V)): Map[K, V]