Search code examples
pythonreactivexrx-py

RxPy: How to create hot observable from external callback and subscribe multiple asynchronous processes?

I have a external service (ExternalDummyService) in which I register a callback. I want to create an observable from that callback and subscribe multiple asynchronous processes.

Full code in pyfiddle: https://pyfiddle.io/fiddle/da1e1d53-2e34-4742-a0b9-07838f2c13df * Note that in the pyfiddle version, the "sleeps" are replaced with "for i in range(10000): foo += i" because sleep does not work properly.

The main code is this:

thread = ExternalDummyService()
external_obs = thread.subject.publish()

external_obs.subscribe(slow_process)
external_obs.subscribe(fast_process)
external_obs.connect()

thread.start()

class ExternalDummyService(Thread):
    def __init__(self):
        self.subject = Subject()

    def run(self):
        for i in range(5):
            dummy_msg = { ... }
            self.subject.on_next(dummy_msg)

def fast_process(msg):
    print("FAST {0} {1}".format(msg["counter"], 1000*(time() - msg["timestamp"])))
    sleep(0.1)

def slow_process(msg):
    print("SLOW {0} {1}".format(msg["counter"], 1000*(time() - msg["timestamp"])))
    sleep(1)

The output I'm getting is this one, with both processes running synchronously and the ExternalDummyService not emitting new values until both processes has finished each execution:

emitting 0
STARTED
SLOW 0 1.0008811950683594
FAST 0 2.0122528076171875
emitting 1
SLOW 1 1.5070438385009766
FAST 1 1.5070438385009766
emitting 2
SLOW 2 0.5052089691162109
FAST 2 0.9891986846923828
emitting 3
SLOW 3 1.0006427764892578
FAST 3 1.0006427764892578
emitting 4
SLOW 4 1.0013580322265625
FAST 4 1.0013580322265625

FINISHED

I would like to get something like this, with the service emitting messages without waiting for the processes to run and the processes running asynchronously:

STARTED
emitting 0
emitting 1
emitting 2
FAST 0 2.0122528076171875
FAST 1 1.5070438385009766
emitting 3
SLOW 0 1.0008811950683594
FAST 2 0.9891986846923828
emitting 4
FAST 3 1.0006427764892578
SLOW 1 1.5070438385009766
FAST 4 1.0013580322265625
SLOW 2 0.5052089691162109
SLOW 3 1.0006427764892578
SLOW 4 1.0013580322265625

FINISHED

I have tried with share(), ThreadPoolScheduler and other I-have-no-idea-what-i'm-doing things.

Thanks!

Solution

  • Using the answer from this question: RxJava concurrency with multiple subscribers and events

    ... I have achieved the desired result with this code:

    optimal_thread_count = cpu_count()
pool_scheduler = ThreadPoolScheduler(optimal_thread_count)

thread = ExternalDummyService()
external_obs = thread.subject.publish()

external_obs \
    .flat_map(lambda msg: Observable.just(msg).subscribe_on(pool_scheduler)) \
    .subscribe(fast_process)

external_obs \
    .flat_map(lambda msg: Observable.just(msg).subscribe_on(pool_scheduler)) \
    .subscribe(slow_process)

external_obs.connect()

thread.start()

    Full version: https://pyfiddle.io/fiddle/20f8871c-48d6-4d6b-b1a4-fdd0a4aa6f95/?m=Saved%20fiddle

    The output is:

    emitting 0
emitting 1
emitting 2
emitting 3
emitting 4
FAST 0 52.629709243774414
FAST 1 51.12814903259277
FAST 2 100.2051830291748
FAST 3 151.2434482574463
SLOW 0 503.0245780944824
SLOW 1 502.0263195037842
FAST 4 548.7725734710693
SLOW 2 551.4400005340576
SLOW 3 652.1098613739014
SLOW 4 1000.3445148468018

    Please, feel free to suggest any improvement.