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]

[adamw]

@halotukozak I don't think it's possible to implement operators which yield more than one flow. It's not clear then what the semantics would be, since we are operating on the level of lazy-evaluated descriptions. After def partition(predicate: A => Boolean): (Flow[A], Flow[A]), running the first flow would run the original flow. But what should happen when running the second flow? Should it run the original once more? Or should there be a shared channel?

What might make sense is implementing sth like def partition(predicate: A => Boolean): (Source[A], Source[A]), that is you get two Channels (eager) which can be received from. If you'd have specific use-cases for these operators, maybe we can come up with the right solutions.

[halotukozak]

@halotukozak I don't think it's possible to implement operators which yield more than one flow. It's not clear then what the semantics would be, since we are operating on the level of lazy-evaluated descriptions. After def partition(predicate: A => Boolean): (Flow[A], Flow[A]), running the first flow would run the original flow. But what should happen when running the second flow? Should it run the original once more? Or should there be a shared channel?

What might make sense is implementing sth like def partition(predicate: A => Boolean): (Source[A], Source[A]), that is you get two Channels (eager) which can be received from. If you'd have specific use-cases for these operators, maybe we can come up with the right solutions.

Nothing particular, just experimenting with Flow in place of the Seq

[adamw]

More operators: #417