@extensionclassmethod(Observable)
def when(cls, *args):

    plans = args[0] if len(args) and isinstance(args[0], list) else list(args)

    def subscribe(observer):
        active_plans = []
        external_subscriptions = {}

        def on_error(err):
            for v in external_subscriptions.values():
                v.on_error(err)
            observer.on_error(err)

        out_observer = Observer(observer.on_next, on_error, 
                                observer.on_completed)

        def deactivate(active_plan):
            active_plans.remove(active_plan)
            if not len(active_plans):
                observer.on_completed()
        try:
            for plan in plans:
                active_plans.append(plan.activate(external_subscriptions, 
                                                  out_observer, deactivate))
        except Exception as ex:
            Observable.throw(ex).subscribe(observer)

        group = CompositeDisposable()
        for join_observer in external_subscriptions.values():
            join_observer.subscribe()
            group.add(join_observer)

        return group

    return AnonymousObservable(subscribe)
Obtain a new event loop.
dataset/ETHPy150Open ReactiveX/RxPY/rx/linq/observable/when.py/when--crossover with--dataset/ETHPy150Open quantmind/pulsar/pulsar/async/access.py/new_event_loop