Concurrent Execution Not Working as Expected with RxPY ThreadPoolScheduler

[vjcspy]

Concurrent Execution Not Working as Expected with RxPY ThreadPoolScheduler

[matiboy]

Hi @vjcspy

For extra information you can refer to @dbrattli 's reply on an older issue

That issue being from v3 times, below is a sample code that should get you the expected parallelism.

def main():
    import time

    import reactivex
    from reactivex import operators as ops
    from reactivex.scheduler import ThreadPoolScheduler

    start = time.time()

    pool_scheduler = ThreadPoolScheduler(5)
    messages = []

    def log(message):
        with threading.Lock():
            messages.append(f"{round(time.time() - start , 1)}: {message}")

    def intense_calculation(value):
        time.sleep(2)
        return f"Computed for {value}"
    
    # Create an Observable
    source = reactivex.from_(["Alpha", "Beta", "Gamma"])
    
    source.pipe(
        ops.flat_map(lambda s: reactivex.from_future(pool_scheduler.executor.submit(intense_calculation, s))),
    ).subscribe(
        on_next=lambda s: log(
            f"Processed {s} on {threading.current_thread().name}"
        ),
        on_error=lambda e: log(f"ERROR {e}"),
        on_completed=lambda: log("Process complete!"),
    )

    
    time.sleep(3)
    log("End")
    print(messages)

Additional issues that discuss this: #67

Hope this helps