Propose shortcut for Iterable<Deferred<T>> → Flow<T> e.g. map { it::await.asFlow() }.merge()

[jablko]

What do you think about adding the following to the library:

/**
 * Creates a flow from a collection of deferreds. Elements are emitted in the order they complete.
 */
fun <T> Iterable<Deferred<T>>.race(): Flow<T> = map { it::await.asFlow() }.merge()

Call it:

• Iterable<Deferred<T>>.merge() like Iterable<Flow<T>>.merge()
• or asCompleted() like Python concurrent.futures.as_completed()
• or my personal favorite, race() like Promise.race()

Currently I use deferreds.map { it::await.asFlow() }.merge().filterNotNull().firstOrNull() to get the first in a collection of deferreds to complete with a non-null value.

The proposed shortcut isn't much shorter, but:

• I think a shortcut for Iterable<Deferred<T>> → Flow<T> has general utility
• it's analogous to other libraries, i.e. Python concurrent.futures.as_completed(), Promise.race()
• I think naming this idiom (map { it::await.asFlow() }.merge()) makes it clearer and easier to find and use?

Unless there's already a better way of doing the same thing?

Previously proposed during the discussion of #171

[lowasser]

Why not do this as

channelFlow {
  for (deferred in this) {
    launch { send(deferred.await()) }
  }
}

[jablko]

Sounds good, I don't know the merits of map { it::await.asFlow() }.merge() vs. channelFlow { ... }, presumably the latter is more efficient?

Another point in favor of a canonical shortcut?

[qwwdfsad]

Could you please elaborate on the usage of such API? E.g. what exact problem you are solving with that?
Could be the case that #1147 or #172 are more relevant

[pacher]

If it gets implemented, I think more idiomatic would be to call it awaitAny() to complement existing awaitAll().

[jablko]

Could you please elaborate on the usage of such API? E.g. what exact problem you are solving with that?

I have a collection of deferreds and I need to get the first one to complete with a non-null value. Backing up, I start with a collection of objects (tax forms) which I evaluate concurrently and I need to get the first occurrence of a specified tax line. Each form returns null if that line wasn't found.

My code looks something like the following (eliding most of it):

class EvalVisitor(file: File, val scope: CoroutineScope) {
  // ...

  // Look up a single line number
  fun FormParser.RangesContext.single(): Deferred<Any?>? {
    val (line, part, form) = visit(this).single()
    // ...
    // First occurrence in the listed forms of the specified line number
    return listOf(yourReturn, schedule1, form428).get(line, part)
  }

  fun Iterable<EvalVisitor?>.get(line: String, part: String? = null): Deferred<Any?> {
    // Evaluate each form concurrently
    val bound = mapNotNull { it?.get(line, part, this@EvalVisitor) }
    // First non-null result
    return scope.async { bound.map { it::await.asFlow() }.merge().filterNotNull().firstOrNull() }
  }

  fun get(line: String, part: String?, caller: EvalVisitor): Deferred<Any?>? {
    // ...
  }
}

In the absence of a shortcut I use bound.map { it::await.asFlow() }.merge().filterNotNull().firstOrNull() to get the first deferred to complete with a non-null value, or alternatively:

  channelFlow {
        for (deferred in bound) {
          launch { send(deferred.await()) }
        }
      }
      .filterNotNull()
      .firstOrNull()

or

  channelFlow {
        for (deferred in bound) {
          launch { deferred.await()?.let(::send) }
        }
      }
      .firstOrNull()

I propose e.g. bound.race().filterNotNull().firstOrNull() modulo what you name the method.

Could be the case that #1147 or #172 are more relevant

Ultimately I start with an Iterable<EvalVisitor?> (list of forms) and I need a Deferred<Any?> (first non-null result). Making use of flows in the middle seemed like the right approach because composable operators (.filterNotNull() and .firstOrNull()) already exist (one could use deferreds.race().firstOrNull(predicate) to get the first deferred to satisfy a generic predicate if needed).

Keeping with flows, I could introduce them before concurrently evaluating the forms vs. after (flowOf(yourReturn, schedule1, form428) vs. map { it::await.asFlow() }), however I don't think this adds any clarity or efficiency?

  flowOf(yourReturn, schedule1, form428)
      // Like in https://github.com/Kotlin/kotlinx.coroutines/issues/1147
      .concurrent { flow -> flow.mapNotNull { it?.get(line, part, this@EvalVisitor) } }
      .merge()
      .filterNotNull()
      .firstOrNull()

vs.

  listOf(yourReturn, schedule1, form428)
      .mapNotNull { it?.get(line, part, this@EvalVisitor) }
      .race()
      .filterNotNull()
      .firstOrNull()