Flow.take() of outer flow not respected with presence of take() on inner flow

[bnorm]

Example code for 1.3.2:

suspend fun main() {
  println(flow<Int> {
    repeat(4) { i ->
      emitAll(flow {
        repeat(4) { j ->
          emit(i * 10 + j)
        }
      }.take(2))
    }
  }.take(4).toList().joinToString(","))
}

Should print 0,1,10,11 but actually prints 0,1,10,11,20,21,30,31. Changing the outer take(4) to take(1) produces 0,10,11,20,21,30,31.

I believe the AbortFlowException of the outer take(4) is being swallowed by the inner take(2). I rewrote the take operator locally and had it track its specific AbortFlowException and rethrow if it did not match and that corrected the problem.

fun <T> Flow<T>.take(count: Int): Flow<T> {
  require(count > 0) { "Requested element count $count should be positive" }
  return flow {
    // Creating the exception eagerly is obviously expensive but it was easy for testing
    val ours = AbortFlowException()
    var consumed = 0
    try {
      collect { value ->
        emit(value)
        if (++consumed == count) {
          throw ours
        }
      }
    } catch (e: AbortFlowException) {
      if (e != ours) throw e
    }
  }
}

There might be errors when combining other operators that use AbortFlowException as it seems to be a common pattern to just ignore this exception when thrown.