concatEager does not unsubscribe from sources

[rossdanderson]

Hi, I have a question regarding the behaviour of concatEager in version 2.2.6.

We were caught out recently by the behaviour of concatEager when the downstream unsubscribes before the upstream sources have completed, namely that the upstream sources are not unsubscribed from when this happens. This left us with a rather nasty memory leak in our application due to some downstream periodical resubscription logic.

Consider the following two tests (apologies for the Kotlin, I can convert to Java on request, but I hope it makes sense)

    @Test
    fun concat() {
        val _publisher1 = PublishProcessor.create<Unit>()
        val _publisher2 = PublishProcessor.create<Unit>()
        val publisher1 = _publisher1.doOnSubscribe { println("1 Subscribed") }.doOnCancel { println("1 Cancelled") }
        val publisher2 = _publisher2.doOnSubscribe { println("2 Subscribed") }.doOnCancel { println("2 Cancelled") }

        assertFalse(_publisher1.hasSubscribers())
        assertFalse(_publisher2.hasSubscribers())

        val testSubscriber = Flowable.concat(Flowable.just(publisher1, publisher2))
            .subscribe()

        assertTrue(_publisher1.hasSubscribers())
        assertFalse(_publisher2.hasSubscribers())

        testSubscriber.dispose()

        assertFalse(_publisher1.hasSubscribers())
        assertFalse(_publisher2.hasSubscribers())
    }

    @Test
    fun concatEager() {
        val _publisher1 = PublishProcessor.create<Unit>()
        val _publisher2 = PublishProcessor.create<Unit>()
        val publisher1 = _publisher1.doOnSubscribe { println("1 Subscribed") }.doOnCancel { println("1 Cancelled") }
        val publisher2 = _publisher2.doOnSubscribe { println("2 Subscribed") }.doOnCancel { println("2 Cancelled") }

        assertFalse(_publisher1.hasSubscribers())
        assertFalse(_publisher2.hasSubscribers())

        val testSubscriber = Flowable.concatEager(Flowable.just(publisher1, publisher2))
            .subscribe()

        assertTrue(_publisher1.hasSubscribers())
        assertTrue(_publisher2.hasSubscribers())

        testSubscriber.dispose()

        assertFalse(_publisher1.hasSubscribers())
        assertFalse(_publisher2.hasSubscribers())
    }

The first test (concat) completes successfully with the following output:

1 Subscribed
1 Cancelled

As we expected, _publisher1 was unsubscribed from when the testSubscriber is disposed.

The second test (concatEager) fails however with the following output:

1 Subscribed
2 Subscribed
2 Cancelled

This doesn't behave as we expected, with the current active publisher _publisher1 not having been unsubscribed from when the testSubscriber is disposed.

Now the wording on the documentation is different - concat:

Returns a Flowable that emits the items emitted by each of the Publishers emitted by the source Publisher, one after the other, without interleaving them.

vs concatEager:

The operator buffers the values emitted by these Publishers and then drains them in order, each one after the previous one completes.

But this was not enough to convey the rather large difference in behaviour with the upstream dispose, to us at least, if this is indeed what it was attempting to convey.

So my questions really are
Is this behaviour expected, or should concatEager dispose all of the upstream sources, including the active one, when it is itself disposed?
If this is the expected behaviour, do you think the documentation should/could be improved here?

[akarnokd]

concatEager is supposed to dispose all sources. I'll investigate it.

[rossdanderson]

Thanks, I appreciate it

[akarnokd]

This is a bug in the operator. I posted a fix in #6405.

[rossdanderson]

👍 That was very quick, thanks again