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

[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.