Job.cancel() doesn't dispose SingleSource called via rx2 await()

[CarsonRedeye]

I have a presenter that implements CoroutineScope and uses a SupervisorJob as the scope job.

Within that presenter, I create a long running network request like this:

launch {
    withContext(Dispatchers.IO) {
        repo.downloadThing(thingId)
    }
}

Then I quickly cancel the job in the onDestroy function of my presenter

fun destroy() {
    supervisorJob.cancel()
}

My SingleSource isn't disposed:

suspend fun downloadThing(thingId: String) {
    service.download(thingId)
                .doOnDispose { Timber.d("DISPOSED") } // This is never called
                .doOnError { Timber.d("ERROR") } // This is called only after the network call completes (leaked)
                .await()
}

Using version 1.3.2

[qwwdfsad]

Could you please provide a self-contained reproducer?
We have both code and tests ensuring that the source is properly disposed when its awaiter is cancelled.
E.g. the following smoke successfully prints "Disposed":

val awaiter = launch { //
    Single.never<Int>()
        .doOnDispose { println("Disposed") }
        .doOnError { println("Error") }
        .await()
}
yield()
awaiter.cancelAndJoin()

Thus either we have a sophisticated bug hidden in some complex interaction (and then we definitely need a reproducer to fix it) or it's just a bug in your testing/application code

[CarsonRedeye]

I am trying to provide a re-producer but I'm not familiar with testing coroutines. For example I don't know what yield() does here. Would it be ok to provide a small android app example?

[qwwdfsad]

Yes, android app should do the trick. Basically, anything that I can just clone & run will work

[CarsonRedeye]

Ok so I've narrowed it down to happening only if Retrofit is used with Rxjava adapter. If it is used with suspend function it cancels correctly

[CarsonRedeye]

Here is a reproduction app. I'm not sure if it is a retrofit problem or an rx2 await problem. The workaround is to call subscribeOn(Schedulers.io()) again on the single that is being called by rx2. https://drive.google.com/open?id=16kK23594rI_6Qa9tiOZBi69n8BEg6pGS

[qwwdfsad]

Thanks, that really helps!
Indeed, this is an interaction on the edge of coroutines cancellation and Rx API. The problem here is that getSomethingSlowly().subscribe() (transitively via await) blocks the current thread before dependency between cancellation of a coroutine and disposal of a subscription is established. Using subsribeOn with any scheduler apart from direct/trampoline should fix your problem.

I have a potential fix, but I am not sure whether we can include it in 1.3.3