zip(o1, o2) doesn't unsubscribe from o1

[ultimate-deej]

A simple example, looks ok

Observable<Integer> o1 = Observable.just(1)
    .doOnCompleted(() -> System.out.println("o1 completed"));

Observable<Integer> o2 = Observable.just(1, 2, 3, 4)
    .doOnCompleted(() -> System.out.println("o2 completed"));

Observable.zip(o1, o2, (integer1, integer2) -> null).subscribe();

Output:

o1 completed

Now swap o1 and o2

Observable<Integer> o1 = Observable.just(1, 2, 3, 4)
    .doOnCompleted(() -> System.out.println("o1 completed"));

Observable<Integer> o2 = Observable.just(1)
    .doOnCompleted(() -> System.out.println("o2 completed"));

Observable.zip(o1, o2, (integer1, integer2) -> null).subscribe();

Output:

o1 completed
o2 completed

[akarnokd]

Zip uses a prefetch of 128 elements by default. Since both sources are synchronous, the second case takes all 4 from the first source, including completion, then moves onto o2 which first emits 1, then its completion completes the whole sequence. Thus, you see both doOnCompleted executed.

The first case, o1 signals 1 and completion, then once o2 signals its 1, it is known o2 won't ever signal more values, thus o2 gets unsubscribed and never allowed to run to completion.

Both are the expected behavior. If you need resource cleanup associated with a sequence, you can't rely on doOnCompleted as sequences may get unsubscribed before that.

[ultimate-deej]

I see, but isn't this implementation details?

[akarnokd]

It is the explanation for what you experienced. The zip javadoc has this clause:

The resulting Observable<R> returned from zip will invoke Observer#onNext
as many times as the number of onNext invocations of the source Observable that emits the fewest
items.

Since zip's behavior seems to be confusing, maybe we could add a clause that zip completes eagerly if one of the sources emits fewer elements than the others and unsubscribes the rest.

[ultimate-deej]

I agree that docs should be updated. Although the new clause you suggested needs to be more accurate:

1. Examples in this issue contradict it
2. If source observables emit equal number of elements (and not fewer), some (which? this one really bothers me) are still going to be unsubscribed.

[akarnokd]

See the proposed documentation changes in #3981.

[ultimate-deej]

Thanks, now I know what to expect from zip. The current implementation still feels unpolished to me, though.