zip operator should complete if a single upstream source completes

[MichaelSchneeberger]

The following observable y ...

import rx
from rx import operators as rxop

y = rx.range(2).pipe(
    rxop.zip(rx.defer(lambda _: y)),
    rxop.merge(rx.just(0)),
    rxop.share(),
)

y.subscribe(print, on_completed=lambda: print('completed'))

... never completes.

My suggestion: The zip operator should complete if any upstream source completed and the corresponding queue is empty.

That is how it is implemented in RxJava.

[MainRo]

I agree that zip should complete in such cases, because no item will be emitted after one source completes. Some applications may use the current behavior to wait for all streams to complete, but probably most people expect it to complete immediately.

[hoc081098]

Hi @MichaelSchneeberger, I created a PR #537