@extensionmethod(Observable)
def group_join(self, right, left_duration_selector, right_duration_selector,
               result_selector):

    left = self

    def subscribe(observer):
        nothing = lambda _: None
        group = CompositeDisposable()
        r = RefCountDisposable(group)
        left_map = OrderedDict()
        right_map = OrderedDict()
        left_id = [0]
        right_id = [0]

        def on_next_left(value):
            s = Subject()

            with self.lock:
                _id = left_id[0]
                left_id[0] += 1
                left_map[_id] = s

            try:
                result = result_selector(value, add_ref(s, r))
            except Exception as e:
                log.error("*** Exception: %s" % e)
                for left_value in left_map.values():
                    left_value.on_error(e)

                observer.on_error(e)
                return

            observer.on_next(result)

            for right_value in right_map.values():
                s.on_next(right_value)

            md = SingleAssignmentDisposable()
            group.add(md)

            def expire():
                if _id in left_map:
                    del left_map[_id]
                    s.on_completed()

                group.remove(md)

            try:
                duration = left_duration_selector(value)
            except Exception as e:
                for left_value in left_map.values():
                    left_value.on_error(e)

                observer.on_error(e)
                return

            def on_error(e):
                for left_value in left_map.values():
                    left_value.on_error(e)

                observer.on_error(e)

            md.disposable = duration.take(1).subscribe(
                nothing,
                on_error,
                expire)

        def on_error_left(e):
            for left_value in left_map.values():
                left_value.on_error(e)

            observer.on_error(e)

        group.add(left.subscribe(on_next_left, on_error_left,
                  observer.on_completed))

        def on_next_right(value):
            with self.lock:
                _id = right_id[0]
                right_id[0] += 1
                right_map[_id] = value

            md = SingleAssignmentDisposable()
            group.add(md)

            def expire():
                del right_map[_id]
                group.remove(md)

            try:
                duration = right_duration_selector(value)
            except Exception as e:
                for left_value in left_map.values():
                    left_value.on_error(e)

                observer.on_error(e)
                return

            def on_error(e):
                with self.lock:
                    for left_value in left_map.values():
                        left_value.on_error(e)

                    observer.on_error(e)

            md.disposable = duration.take(1).subscribe(
                nothing,
                on_error,
                expire)

            with self.lock:
                for left_value in left_map.values():
                    left_value.on_next(value)

        def on_error_right(e):
            for left_value in left_map.values():
                left_value.on_error(e)

            observer.on_error(e)

        group.add(right.subscribe(on_next_right, on_error_right))
        return r
    return AnonymousObservable(subscribe)
Returns an initialized memcache client.  Used by the constructor.
dataset/ETHPy150Open ReactiveX/RxPY/rx/linq/observable/groupjoin.py/group_join--crossover with--dataset/ETHPy150Open jojoin/cutout/cutout/cache/memcachedcache.py/MemcachedCache.import_preferred_memcache_lib