Job.cancel does not dispose Rx Observable when using consumeEach

[mhernand40]

As I'm looking to convert my Singles, Maybes, and Completables into suspend functions, while still relying on RxJava's Observables, I was hoping that CoroutineScope could manage the subscription to my Observable streams by relying on the consumeEach extension function. However, it turns out that when the Job gets cancelled, although no more values will be "consumed" by the underlying Channel, the Observable itself will not get disposed. This could potentially leak the Observable.

Sample Repro:

@Test
fun repro() {
    val observable = Observable.interval(1L, TimeUnit.SECONDS)
        .doOnNext { println("onNext received $it") }
        .doOnDispose { println("Observable disposed!") }

    val job = GlobalScope.launch {
        observable.consumeEach{
            println("Value $it consumed")
        }
    }

    Thread.sleep(3000L)
    job.cancel()
    Thread.sleep(3000L)
}

When I run this test I get the following result:

onNext received 0
Value 0 consumed
onNext received 1
Value 1 consumed
onNext received 2
onNext received 3
onNext received 4
onNext received 5

[elizarov]

That is a bug. That's for the report.