python - RxPy:外部コールバックからホットオブザーバブルを作成し、複数の非同期プロセスをサブスクライブする方法は?

Question

コールバックを登録する外部サービス (ExternalDummyService) があります。そのコールバックからオブザーバブルを作成し、複数の非同期プロセスをサブスクライブしたいと考えています。

pyfiddle の完全なコード: https://pyfiddle.io/fiddle/da1e1d53-2e34-4742-a0b9-07838f2c13df * pyfiddle バージョンでは、「スリープ」が「for i in range(10000): foo +」に置き換えられていることに注意してください。 = i" というのは、スリープが正常に機能しないためです。

主なコードは次のとおりです。

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)

私が得ている出力は次のとおりです。両方のプロセスが同期的に実行されており、両方のプロセスが各実行を完了するまで、ExternalDummyService は新しい値を発行しません。

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

プロセスが実行されるのを待たずにサービスがメッセージを発行し、プロセスが非同期で実行されるようにして、次のようなものを取得したいと思います。

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

私はshare()、ThreadPoolScheduler、およびその他の私が何をしているのかわからないことを試しました。

ありがとう！