<commit_before>from six import add_metaclass

from rx.observable import Observable
from rx.anonymousobservable import AnonymousObservable
from rx.disposables import CompositeDisposable, SingleAssignmentDisposable
from rx.concurrency import Scheduler, immediate_scheduler
from rx.internal import ExtensionMethod

@add_metaclass(ExtensionMethod)
class ObservableMerge(Observable):
    def __init__(self, subscribe):
        self.merge = self.__merge

    def __merge(self, max_concurrent_or_other):
        """Merges an observable sequence of observable sequences into an
        observable sequence, limiting the number of concurrent subscriptions
        to inner sequences. Or merges two observable sequences into a single
        observable sequence.

        1 - merged = sources.merge(1)
        2 - merged = source.merge(other_source)

        max_concurrent_or_other [Optional] Maximum number of inner observable
            sequences being subscribed to concurrently or the second
            observable sequence.

        Returns the observable sequence that merges the elements of the inner
        sequences."""

        if not isinstance(max_concurrent_or_other, int):
            return Observable.merge(max_concurrent_or_other)

        sources = self

        def subscribe(observer):
            active_count = [0]
            group = CompositeDisposable()
            is_stopped = [False]
            q = []

            def subscribe(xs):
                subscription = SingleAssignmentDisposable()
                group.add(subscription)

                def on_completed():
                    group.remove(subscription)
                    if q.length > 0:
                        s = q.shift()
                        subscribe(s)
                    else:
                        active_count[0] -= 1
                        if is_stopped[0] and active_count[0] == 0:
                            observer.on_completed()

                subscription.disposable = xs.subscribe(observer.on_next, observer.on_error, on_completed)

            def on_next(inner_source):
                if active_count[0] < max_concurrent_or_other:
                    active_count[0] += 1
                    subscribe(inner_source)
                else:
                    q.append(inner_source)

            def on_completed():
                is_stopped[0] = True
                if active_count[0] == 0:
                    observer.on_completed()

            group.add(sources.subscribe(on_next, observer.on_error, on_completed))
            return group
        return AnonymousObservable(subscribe)

    @classmethod
    def merge(cls, *args):
        """Merges all the observable sequences into a single observable
        sequence. The scheduler is optional and if not specified, the
        immediate scheduler is used.

        1 - merged = rx.Observable.merge(xs, ys, zs)
        2 - merged = rx.Observable.merge([xs, ys, zs])
        3 - merged = rx.Observable.merge(scheduler, xs, ys, zs)
        4 - merged = rx.Observable.merge(scheduler, [xs, ys, zs])

        Returns the observable sequence that merges the elements of the
        observable sequences."""

        if not args[0]:
            scheduler = immediate_scheduler
            sources = args[1:]
        elif isinstance(args[0], Scheduler):
            scheduler = args[0]
            sources = args[1:]
        else:
            scheduler = immediate_scheduler
            sources = args[:]

        if isinstance(sources[0], list):
            sources = sources[0]

        return Observable.from_array(sources, scheduler).merge_observable()

    def merge_all(self):
        """Merges an observable sequence of observable sequences into an
        observable sequence.

        Returns the observable sequence that merges the elements of the inner
        sequences."""

        sources = self

        def subscribe(observer):
            group = CompositeDisposable()
            is_stopped = False
            m = SingleAssignmentDisposable()
            group.add(m)

            def on_next(inner_source):
                inner_subscription = SingleAssignmentDisposable()
                group.add(inner_subscription)

                inner_source = Observable.from_future(inner_source)

                def on_next(x):
                    observer.on_next(x)

                def on_completed():
                    group.remove(inner_subscription)
                    if is_stopped and group.length == 1:
                        observer.on_completed()

                inner_subscription.disposable = inner_source.subscribe(on_next, observer.on_error, on_completed)

            def on_completed():
                is_stopped = True
                if len(group) == 1:
                    observer.on_completed()

            m.disposable = sources.subscribe(on_next, observer.on_error, on_completed)
            return group

        return AnonymousObservable(subscribe)

    def merge_observable(self):
        """Merges an observable sequence of observable sequences into an
        observable sequence.

        Returns the observable sequence that merges the elements of the inner
        sequences.
        """
        sources = self

        def subscribe(observer):
            m = SingleAssignmentDisposable()
            group = CompositeDisposable()
            is_stopped = [False]
            group.add(m)

            def on_next(inner_source):
                inner_subscription = SingleAssignmentDisposable()
                group.add(inner_subscription)

                inner_source = Observable.from_future(inner_source)

                def on_complete():
                    group.remove(inner_subscription)
                    if is_stopped[0] and group.length == 1:
                        observer.on_completed()

                disposable = inner_source.subscribe(
                    observer.on_next,
                    observer.on_error,
                    on_complete)

                inner_subscription.disposable = disposable

            def on_complete():
                is_stopped[0] = True
                if group.length == 1:
                    observer.on_completed()

            m.disposable = sources.subscribe(on_next, observer.on_error, on_complete)
            return group

        return AnonymousObservable(subscribe)
<commit_msg>Bugfix for merge<commit_after>from six import add_metaclass

from rx.observable import Observable
from rx.anonymousobservable import AnonymousObservable
from rx.disposables import CompositeDisposable, SingleAssignmentDisposable
from rx.concurrency import Scheduler, immediate_scheduler
from rx.internal import ExtensionMethod

@add_metaclass(ExtensionMethod)
class ObservableMerge(Observable):
    def __init__(self, subscribe):
        self.merge = self.__merge

    def __merge(self, max_concurrent_or_other):
        """Merges an observable sequence of observable sequences into an
        observable sequence, limiting the number of concurrent subscriptions
        to inner sequences. Or merges two observable sequences into a single
        observable sequence.

        1 - merged = sources.merge(1)
        2 - merged = source.merge(other_source)

        max_concurrent_or_other [Optional] Maximum number of inner observable
            sequences being subscribed to concurrently or the second
            observable sequence.

        Returns the observable sequence that merges the elements of the inner
        sequences."""

        if not isinstance(max_concurrent_or_other, int):
            return Observable.merge(self, max_concurrent_or_other)

        sources = self

        def subscribe(observer):
            active_count = [0]
            group = CompositeDisposable()
            is_stopped = [False]
            q = []

            def subscribe(xs):
                subscription = SingleAssignmentDisposable()
                group.add(subscription)

                def on_completed():
                    group.remove(subscription)
                    if q.length > 0:
                        s = q.shift()
                        subscribe(s)
                    else:
                        active_count[0] -= 1
                        if is_stopped[0] and active_count[0] == 0:
                            observer.on_completed()

                subscription.disposable = xs.subscribe(observer.on_next, observer.on_error, on_completed)

            def on_next(inner_source):
                if active_count[0] < max_concurrent_or_other:
                    active_count[0] += 1
                    subscribe(inner_source)
                else:
                    q.append(inner_source)

            def on_completed():
                is_stopped[0] = True
                if active_count[0] == 0:
                    observer.on_completed()

            group.add(sources.subscribe(on_next, observer.on_error, on_completed))
            return group
        return AnonymousObservable(subscribe)

    @classmethod
    def merge(cls, *args):
        """Merges all the observable sequences into a single observable
        sequence. The scheduler is optional and if not specified, the
        immediate scheduler is used.

        1 - merged = rx.Observable.merge(xs, ys, zs)
        2 - merged = rx.Observable.merge([xs, ys, zs])
        3 - merged = rx.Observable.merge(scheduler, xs, ys, zs)
        4 - merged = rx.Observable.merge(scheduler, [xs, ys, zs])

        Returns the observable sequence that merges the elements of the
        observable sequences."""

        if not args[0]:
            scheduler = immediate_scheduler
            sources = args[1:]
        elif isinstance(args[0], Scheduler):
            scheduler = args[0]
            sources = args[1:]
        else:
            scheduler = immediate_scheduler
            sources = args[:]

        if isinstance(sources[0], list):
            sources = sources[0]

        return Observable.from_array(sources, scheduler).merge_observable()

    def merge_all(self):
        """Merges an observable sequence of observable sequences into an
        observable sequence.

        Returns the observable sequence that merges the elements of the inner
        sequences."""

        sources = self

        def subscribe(observer):
            group = CompositeDisposable()
            is_stopped = False
            m = SingleAssignmentDisposable()
            group.add(m)

            def on_next(inner_source):
                inner_subscription = SingleAssignmentDisposable()
                group.add(inner_subscription)

                inner_source = Observable.from_future(inner_source)

                def on_next(x):
                    observer.on_next(x)

                def on_completed():
                    group.remove(inner_subscription)
                    if is_stopped and group.length == 1:
                        observer.on_completed()

                inner_subscription.disposable = inner_source.subscribe(on_next, observer.on_error, on_completed)

            def on_completed():
                is_stopped = True
                if len(group) == 1:
                    observer.on_completed()

            m.disposable = sources.subscribe(on_next, observer.on_error, on_completed)
            return group

        return AnonymousObservable(subscribe)

    def merge_observable(self):
        """Merges an observable sequence of observable sequences into an
        observable sequence.

        Returns the observable sequence that merges the elements of the inner
        sequences.
        """
        sources = self

        def subscribe(observer):
            m = SingleAssignmentDisposable()
            group = CompositeDisposable()
            is_stopped = [False]
            group.add(m)

            def on_next(inner_source):
                inner_subscription = SingleAssignmentDisposable()
                group.add(inner_subscription)

                inner_source = Observable.from_future(inner_source)

                def on_complete():
                    group.remove(inner_subscription)
                    if is_stopped[0] and group.length == 1:
                        observer.on_completed()

                disposable = inner_source.subscribe(
                    observer.on_next,
                    observer.on_error,
                    on_complete)

                inner_subscription.disposable = disposable

            def on_complete():
                is_stopped[0] = True
                if group.length == 1:
                    observer.on_completed()

            m.disposable = sources.subscribe(on_next, observer.on_error, on_complete)
            return group

        return AnonymousObservable(subscribe)

{'fraction_non_alphanumeric': 0.06822276180778569, 'fraction_numerical': 0.005169577817811546, 'mean_word_length': 1.8, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4483311', 'n_tokens_mistral': 3289, 'n_tokens_neox': 3003, 'n_words': 821}