[2.2.6] window() operator leaks subscriptions

[vganin]

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.

[akarnokd]

The operator will not dispose the inner windows if you dispose the outer sequence of windows. You should not ignore the inner windows produced by the operator. Most use cases have flatMap to merge the windows back which will dispose all of them as you expected.

[akarnokd]

The printout is certainly odd. I'll investigate further.