flatMap keeps its selector alive beyond inner observable completion

[nikolaykasyanov]

MergeSinkIter retains MergeSink, or to be precise FlatMapSink, that retains the selector.

It's a bit similar to #2604, however I haven't yet figured out how to address it.

flatMapFirst and flatMapLast should be prone to the same issue.

[danielt1263]

This is a bogus test. The flatMap should not complete or free any resources in this use case.

Keep in mind, the Observable returned by flatMap does not emit a completion event until its source Observable and all the Observables it has created emit completion events. In this use case, the inner Observable never completes.

[nikolaykasyanov]

I respectfully disagree, in my opinion resources of the inner observable shouldn't live longer than the subscription to it (which ends with its completion, not with the completion of the observable resulting from flatMap). In the light of your argument, I wonder how #2604 is different.

[nikolaykasyanov]

In this use case, the inner Observable never completes.

If it did, I would have to check if object deallocated some time between the inner observable completion and the outer observable completion instead of just in the end of the test case, which would seem like a a less readable test.

[nikolaykasyanov]

@danielt1263 my apologies, the test was indeed bogus, the object should be captured by the inner observable, not by flatMap's closure. I pushed a fix.

[danielt1263]

Note that this test, which is more like the Completable test you referenced, passes...

func testFlatMap_Complete_OuterNotComplete_DoesNotKeepSelectorAlive() {
    let scheduler = TestScheduler(initialClock: 0)

    let xs = scheduler.createHotObservable([
        .next(100, 1),
        .completed(200)
    ])

    var object = Optional.some(TestObject())

    let disposable =  xs
        .do(onCompleted: { [object] in
            _ = object // this is happening on completion now instead of inside the flatMap
        })
        .flatMap { _ in
            Observable<Never>.never()
        }
        .subscribe()
    defer {
        disposable.dispose()
    }

    scheduler.advanceTo(300) // we have to advance the scheduler
    weak var weakObject = object
    object = nil

    XCTAssertNil(weakObject)
}

[nikolaykasyanov]

@danielt1263 ~this does not pass though:~ nvm it does, no more late evening PRs for me 😞 Thanks for your patience.

[nikolaykasyanov]

@danielt1263 however, getting back to the original test, when the inner one completes, we know that the selector closure will no longer be called, don't we?

[danielt1263]

Actually no, we don't know that the selector will no longer be called. If there is a new subscription to the flatMap's Observable, then the flatMap will resubscribe to its source which will produce a new set of elements. So the FlatMap object needs to retain that closure.

[nikolaykasyanov]

@danielt1263 yes, but particular subscription's FlatMapSink doesn't have to.

[danielt1263]

The FlatMapSink represents the subscription itself. When you call subscribe the Sink will exist until either a completion event happens or the subscription is disposed. In your test, you don't dispose of the subscription until after the assertion and as we have already shown, the subscription (correctly) never completes...

[nikolaykasyanov]

@danielt1263 can't subscriptions "shed" no longer necessary pieces, like the selector closure upon inner subscription's completion in this example, is it a design choice?

[danielt1263]

I guess it could null out the closure at that point, but that seems like a needless optimization.