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

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

    result_selector = args.pop()
    NO_VALUE = object()

    def subscribe(observer):

        def subscribe_all(parent, *children):

            values = [NO_VALUE for _ in children]

            def subscribe_child(i, child):
                subscription = SingleAssignmentDisposable()
                def on_next(value):
                    with parent.lock:
                        values[i] = value
                subscription.disposable = child.subscribe(
                    on_next, observer.on_error)
                return subscription

            parent_subscription = SingleAssignmentDisposable()
            def on_next(value):
                with parent.lock:
                    if NO_VALUE not in values:
                        try:
                            result = result_selector(value, *values)
                        except Exception as error:
                            observer.on_error(error)
                        else:
                            observer.on_next(result)
            parent_subscription.disposable = parent.subscribe(
                on_next, observer.on_error, observer.on_completed)

            return listify_args(
                parent_subscription,
                *(subscribe_child(*a) for a in enumerate(children))
            )

        return CompositeDisposable(subscribe_all(*args))
    return AnonymousObservable(subscribe)
Merges the specified observable sequences into one observable sequence
dataset/ETHPy150Open ReactiveX/RxPY/rx/linq/observable/withlatestfrom.py/with_latest_from