WebSocketとRxPyの連携

rx-ws.py

import json
import threading
from websocket import create_connection
from rx.subjects import Subject
from rx.concurrency import ThreadPoolScheduler

def observable():
    
    # ThreadPool
    pool_scheduler = ThreadPoolScheduler(1)
    stream = Subject()

    # スレッドプロセス
    def ws(stream):
        ws = create_connection('ws://{domain}/messages'.format(domain = 'domain'))
        while True:
            stream.on_next(json.loads(ws.recv()))

    th = threading.Thread(target=ws, name='ws', args=(stream,))
    th.start()

    return stream.subscribe_on(pool_scheduler)