tryrx.py

#!/usr/bin/env python3

from rx import Observable, Observer, AnonymousObservable
from time import sleep
from rx.concurrency import NewThreadScheduler
from concurrent.futures import ThreadPoolExecutor

from rx.internal import extensionmethod

@extensionmethod(Observable)
def repeat_letter(self, letter):
    source = self
    def subscribe(observer):
        def on_next(count):
            observer.on_next(letter * count)

        def on_completed():
            observer.on_completed()

        return source.subscribe(on_next, observer.on_error, on_completed)
    return AnonymousObservable(subscribe)

import random

def shuffle_range(min, max):
    sl = list(range(min, max))
    random.shuffle(sl)
    return sl

if __name__ == "__main__":

    def heavy_stuff(i):
        print("Begining huge work for %i" % i)
        sleep(i)
        print("Finishing huge work for %i" % i)
        return i


    with ThreadPoolExecutor(5) as executor:
        integers = Observable.from_iterable(iter(shuffle_range(1, 10))).flat_map(lambda i: executor.submit(heavy_stuff, i)).repeat_letter('a')
        for integer in integers.to_blocking():
            print(integer)