Add a Flow#handle?

[He-Pin]

When I first learn the Kotlin's Flow api, I think it do has some great elgant.

public fun interface FlowCollector<in T> {

    /**
     * Collects the value emitted by the upstream.
     * This method is not thread-safe and should not be invoked concurrently.
     */
    public suspend fun emit(value: T)
}

Can we get something like it? I think with the AsyncCallback we can.

  trait FlowEmitter[T] {
    def complete():Unit
    def emit( nextValue: T):Unit //at most one?
    def emitMutliValue(nextValues:T.*) //at most one?
    def fail(ex: Throwable):Unit
  }

  def handle[T](f:(Out, FlowEmitter[T]) => Unit): Repr[T] = ???
  
  //implement filter with handle 
  def filter[T](f: Out => Boolean) = handle((out, emitter) =>{
    if(f(out)) {
      emitter.emit(out)
    } 
  })

Or something like it, Have not think much deeper about it yet.

[johanandren]

I personally do not like that imperative API style much, but it seems very close to what mapConcat can do in a more functional style, except that it doesn't have a way to complete/cancel without composition (for example emitting some special value that you can takeWhile on downstream).

Perhaps somewhat related to previous discussions a middle ground between custom graph stage and using predefined operators a couple of times before, some discussion and ideas here: #24944