Experiment implementing amb/race operator using flow and without directly using channels

amb_using_jobs_to_cancel_racers_and_emitting_failures.kt

@OptIn(FlowPreview::class)
fun <A> Flow<A>.ambUsingJobsToCancelRacersAndEmittingFailures(
    vararg others: Flow<A>
): Flow<Either<Throwable, A>> = flow {
    val parentJob = currentCoroutineContext()[Job]!!
    (listOf(this@ambUsingJobsToCancelRacersAndEmittingFailures) + others).foldIndexed(
        Pair(emptyList<Job>(), emptyList<Flow<Pair<Int, Either<Throwable, A>>>>())
    ) { index, acc, curr ->
        val job = Job(parentJob)
        val flow = curr.map { Pair(index, it.right() as Either<Throwable, A>) }
            .catch { emit(Pair(index, it.left() as Either<Throwable, A>)) }
            .onCompletion { println("completion $index $it") }
            .shareIn(CoroutineScope(job), SharingStarted.WhileSubscribed())
        Pair(acc.first + job, acc.second + flow)
    }.let { (jobs, flows) ->
        var winnerIndex: Int? = null
        flows.asFlow()
            .flatMapMerge(concurrency = flows.size) { it }
            .collect { (index, item) ->
                if (winnerIndex == null) {
                    item.fold(ifLeft = {
                        println("cancelling $index")
                        jobs[index].cancelAndJoin()
                    }, ifRight = {
                        winnerIndex = index
                        println("winnerIndex $winnerIndex")
                        jobs.forEachIndexed { i, job ->
                            if (!job.isCancelled && i != winnerIndex) {
                                println("cancelling $i")
                                job.cancelAndJoin()
                            }
                        }
                    })
                }
                emit(item)
                jobs.forEachIndexed { i, job ->
                    println("job cancelled $i ${job.isCancelled}")
                    println("job completed $i ${job.isCompleted}")
                }
            }
        println("done!")
    }
}

val e1 = Exception("Bang!")
val f1 = flow<Int> { throw e1 }.onStart { delay(50) }
val f2 = flowOf(1, 2, 3).onStart { delay(100) }
val f3 = flowOf(10, 20, 30).onStart { delay(150) }
val f4 = f1.ambUsingJobsToCancelRacersAndEmittingFailures(f2, f3)
f4.collect {  // Suspends indefinitely, even after emitting all items in the winning flow.
    println(it)
}

Output

completion 0 null
cancelling 0
Either.Left(java.lang.Exception: Bang!)
job cancelled 0 true
job completed 0 true
job cancelled 1 false
job completed 1 false
job cancelled 2 false
job completed 2 false
completion 1 null
winnerIndex 1
cancelling 2
completion 2 kotlinx.coroutines.JobCancellationException: Job was cancelled; job=JobImpl{Cancelling}@97269b4
Either.Right(1)
job cancelled 0 true
job completed 0 true
job cancelled 1 false
job completed 1 false
job cancelled 2 true
job completed 2 true
Either.Right(2)
job cancelled 0 true
job completed 0 true
job cancelled 1 false
job completed 1 false
job cancelled 2 true
job completed 2 true
Either.Right(3)
job cancelled 0 true
job completed 0 true
job cancelled 1 false
job completed 1 false
job cancelled 2 true
job completed 2 true

Then f4.collect suspends indefinitely.