[2.2.6] window() operator leaks subscriptions #6397
Description
Hi,
I believe I encountered window operator leaking observable subscriptions. Here is a small test to reproduce the issue:
fun main(args: Array<String>) {
observable(1).window(observable(2).filter { it == 1 }, Function { _: Int -> observable(3).filter { it == 1 } }, 1)
.subscribe()
.dispose()
}
private fun observable(n: Int): Observable<Int> {
return Observable.create<Int> { emitter ->
println("Subscribed! $n")
emitter.onNext(1)
emitter.onNext(2)
emitter.onNext(3)
emitter.setCancellable { println("Cancelled! $n") }
}
}
The output is:
Subscribed! 2
Subscribed! 3
Cancelled! 3
Subscribed! 1
From the output you can see that cancellables for observable(1) and observable(2) are never called.
This may be severe if cancellables are used to dispose some (possibly) critical resources upon subscription disposal.