[PyRx Example2] #Python

PyRx

import asyncio
import rx
import rx.operators as op
import time
from rx.scheduler import ThreadPoolScheduler, NewThreadScheduler

scheduler = ThreadPoolScheduler(2)
scheduler = NewThreadScheduler()

def emitter(observer: rx.typing.Observer, _):
    observer.on_next('a')
    time.sleep(.2)
    observer.on_next('b')
    observer.on_completed()

source2 = rx.create(emitter)

async def main(loop):
    x = source2.subscribe(
        on_next = lambda i: print("2: {0}".format(i)),
        on_error = lambda e: print("Error Occurred: {0}".format(e)),
        on_completed = lambda: print("2 Done!"),
        # scheduler=scheduler,
    )
    await source2


loop = asyncio.get_event_loop()
loop.run_until_complete(main(loop))
loop.close()
print('end')