merge() does not serialize on_next() calls

[vermaport]

Similar to ReactiveX/RxJava#200, it seems like merge() does not serialize emissions if its sources are emitting on different threads. Is there an RxPy equivalent to SynchronizedObserver to work around this?

I translated the unit test written for this issue, and it fails on RxPy 1.6.1 (latest in PIP).

ReactiveX/RxJava@effc08d#diff-e28fd5678569ef71dfc9e835c73a5c36R42

from threading import Event, Lock

from rx import Observable
from rx.concurrency import EventLoopScheduler
from rx.core import ObservableBase


class TestAsynchronousObservable(ObservableBase):
    def __init__(self, scheduler=None):
        super(TestAsynchronousObservable, self).__init__()
        self.scheduler = scheduler or EventLoopScheduler()
        self.on_next_being_sent = Event()

    def hello(self, scheduler, observer):
        self.on_next_being_sent.set()
        observer.on_next("hello")
        observer.on_completed()

    def _subscribe_core(self, observer):
        self.scheduler.schedule(self.hello, observer)


counter_lock = Lock()
total_counter = 0
concurrent_counter = 0
unblock = Event()
completed = Event()


def on_next(item):
    global counter_lock
    global total_counter
    global concurrent_counter
    global unblock

    with counter_lock:
        total_counter += 1
        concurrent_counter += 1

    unblock.wait()

    concurrent_counter -= 1


o1 = TestAsynchronousObservable()
o2 = TestAsynchronousObservable()
m = Observable.merge(o1, o2).subscribe(on_next, on_completed=completed.set)

# Wait for both Observables to send.
o1.on_next_being_sent.wait()
o2.on_next_being_sent.wait()

# One of the Observables should be blocked.
assert concurrent_counter == 1  # <-- Asserts here since merge() does not serialize the on_next() calls.

# Release it so it can finish.
unblock.set()

# Wait for the merge to complete.
completed.wait()

assert total_counter == 2
assert concurrent_counter == 0

[lock]

This thread has been automatically locked since there has not been any recent activity after it was closed. Please open a new issue for related bugs.