Using withLatestFrom completes without values if other publisher is delayed

[ivansenic]

Version: 3.0.12

I am not sure if this is per-design and specification (could not find this), but is it really expected that:

      Flowable<Integer> integerFlowable = Flowable.just(1, 2, 3);
      Flowable<Integer> delayedFlowable = Flowable.just(10).delay(1, TimeUnit.SECONDS);

      integerFlowable.withLatestFrom(delayedFlowable, Integer::sum)
              .test()
              .assertValueCount(3)
              .assertComplete();

fails with:

java.lang.AssertionError: Value counts differ; expected: 3 but was: 0 (latch = 0, values = 0, errors = 0, completions = 1)
Expected :3
Actual   :0 (latch = 0, values = 0, errors = 0, completions = 1)
<Click to see difference>

If it is, then Javadoc needs to be improved, it is not stated anywhere that this will complete with no values of the other publisher is emitting item after a delay. Of course this is a simple example, but you get the point. If these would be two database calls and second comes emits after the first one, you get nothing? 😕

[akarnokd]

It is by design. If the upstream ends before any of the other sources had a chance to signal, the outcome is an empty sequence.

[ivansenic]

Thanks @akarnokd. I was checking the PR you made and you said this behavior is unlike the combineLatest, but in the test I did I also get similar error for:

      Flowable<Integer> integerFlowable = Flowable.just(1, 2, 3);
      Flowable<Integer> delayedFlowable = Flowable.just(10).delay(1, TimeUnit.SECONDS);

      Flowable.combineLatest(integerFlowable, delayedFlowable, Integer::sum)
              .test()
              .assertValueCount(3)
              .assertComplete();

java.lang.AssertionError: Value counts differ; expected: 3 but was: 0 (latch = 1, values = 0, errors = 0, completions = 0)
Expected :3
Actual   :0 (latch = 1, values = 0, errors = 0, completions = 0)

OK here latch is reported as 1 and not the completion. So is this also expected per design?

[akarnokd]

Since you run with async flows, you have to use .awaitDone(5, TimeUnit.SECONDS) before asserting to let the code finish.

Also I don't think combineLatest would fit your intentions. integerFlowable would rush to "3" and you'd only get one combination of "3-10".

If you want all combinations, such as "1-10", "2-10", "3-10", you'll need concatMap and cached secondary flow:

Flowable<Integer> integerFlowable = Flowable.just(1, 2, 3);
Flowable<Integer> delayedFlowable = Flowable.just(10).delay(1, TimeUnit.SECONDS).cache();

integerFlowable.concatMap(main -> delayedFlowable.map(other -> Integer.sum(main, other)))
              .test()
              .awaitDone(5, TimeUnit.SECONDS)
              .assertResult(11, 12, 13);

[ivansenic]

Thanks a lot @akarnokd, very informative. 👍