def test_repeat_observable_repeat_count_throws(self):
        scheduler1 = TestScheduler()
        xs = Observable.return_value(1, scheduler1).repeat(3)
        xs.subscribe(lambda x: _raise('ex'))

        with self.assertRaises(RxException):
            scheduler1.start()

        scheduler2 = TestScheduler()
        ys = Observable.throw_exception('ex1', scheduler2).repeat(3)
        ys.subscribe(on_error=lambda ex: _raise('ex2'))

        with self.assertRaises(RxException):
            scheduler2.start()

        scheduler3 = TestScheduler()
        zs = Observable.return_value(1, scheduler3).repeat(100)
        d = zs.subscribe(on_completed=lambda: _raise('ex3'))

        scheduler3.schedule_absolute(10, lambda sc, st: d.dispose())
        scheduler3.start()

        xss = Observable.create(lambda o: _raise('ex4')).repeat(3)
        with self.assertRaises(RxException):
            xss.subscribe()

    def to_future(source: Observable) -> Future:
        """Converts an existing observable sequence to a Future.

        Example:
            future = rx.return_value(42).pipe(ops.to_future(asyncio.Future))

        Args:
            future_ctor: [Optional] The constructor of the future.

        Returns:
            A future with the last value from the observable sequence.
        """

        has_value = []

        def on_next(value):
            has_value.append(value)

        def on_error(err):
            future.set_exception(err)

        def on_completed():
            if has_value:
                future.set_result(has_value.pop())

        source.subscribe_(on_next, on_error, on_completed)

        # No cancellation can be done
        return future

    def subscribe(observer):
        try:
            result = observable_factory()
        except Exception as ex:
            return Observable.throw_exception(ex).subscribe(observer)

        result = Observable.from_future(result)
        return result.subscribe(observer)

def mode(source: Observable) -> Observable:
    """
    Returns the most frequently emitted value (or "values" if they have the same number of occurrences).
    The sequence must be finite.
    """
    return source.group_by(lambda v: v) \
        .flat_map(lambda grp: grp.count().map(lambda ct: (grp.key, ct))) \
        .to_sorted_list(lambda t: t[1], reverse=True) \
        .flat_map(lambda l: Observable.from_(l).take_while(lambda t: t[1] == l[0][1])) \
        .map(lambda t: t[0])

    def test_zip_never_never(self):
        scheduler = TestScheduler()
        o1 = Observable.never()
        o2 = Observable.never()

        def create():
            return o1.zip(o2, lambda x, y: x + y)

        results = scheduler.start(create)
        results.messages.assert_equal()

 def create():
     import sys
     sys.setrecursionlimit(1000)
     def predicate(x):
         n[0] += 1
         return n[0] < 1000
     def subscribe(o):
         o.on_next(1)
         o.on_completed()
         return lambda: None
     return Observable.while_do(predicate, Observable.create(subscribe))

    def test_for_each_index_some_data(self):
        lstX = []
        lstI = []

        def action(x, i):
            lstX.append(x)
            lstI.append(i)

        Observable.range(10, 10).to_blocking().for_each(action)
        assert(lstX == [x for x in range(10, 20)])
        assert(lstI == [x for x in range(10)])

    def test_for_each_index_return(self):
        lstX = []
        lstI = []

        def action(x, i):
            lstX.append(x)
            lstI.append(i)

        Observable.return_value(42).to_blocking().for_each(action)
        assert(lstX == [42])
        assert(lstI == [0])

def variance(source: ObservableBase) -> ObservableBase:
    """
    Returns the statistical variance of the numerical emissions.
    The sequence must be finite.
    """
    squared_values = source.to_list() \
        .flat_map(lambda l: Observable.from_(l).average().flat_map(lambda avg: Observable.from_(l).map(lambda i: i - avg))) \
        .map(lambda i: i * i) \
        .publish() \
        .auto_connect(2)

    return Observable.zip(squared_values.sum(), squared_values.count(), lambda sum, ct: sum / (ct - 1))

    def test_when_never_never(self):
        scheduler = TestScheduler()
        xs = Observable.never()
        ys = Observable.never()

        def create():
            def selector(x, y):
                return x + y
            return Observable.when(xs.and_(ys).then_do(selector))

        results = scheduler.start(create)
        results.messages.assert_equal()

    def test_return_observer_throws(self):
        scheduler1 = TestScheduler()
        xs = Observable.return_value(1, scheduler1)
        xs.subscribe(lambda x: _raise('ex'))

        self.assertRaises(RxException, scheduler1.start)

        scheduler2 = TestScheduler()
        ys = Observable.return_value(1, scheduler2)
        ys.subscribe(lambda x: x, lambda ex: ex, lambda: _raise('ex'))

        self.assertRaises(RxException, scheduler2.start)

def do_while(self, condition):
    """Repeats source as long as condition holds emulating a do while loop.

    Keyword arguments:
    condition -- {Function} The condition which determines if the source
        will be repeated.

    Returns an observable {Observable} sequence which is repeated as long
    as the condition holds.
    """

    return Observable.concat([self, Observable.while_do(condition, self)])

def repeat(self, repeat_count=None):
    """Repeats the observable sequence a specified number of times. If the
    repeat count is not specified, the sequence repeats indefinitely.

    1 - repeated = source.repeat()
    2 - repeated = source.repeat(42)

    Keyword arguments:
    repeat_count -- Number of times to repeat the sequence. If not
        provided, repeats the sequence indefinitely.

    Returns the observable sequence producing the elements of the given
    sequence repeatedly."""

    return Observable.defer(lambda: Observable.concat(Enumerable.repeat(self, repeat_count)))

    def test_many_select_law_1(self):
        xs = Observable.range(1, 0)

        left = xs.many_select(lambda x: x.first())
        right = xs

        left.sequence_equal(right).first().subscribe(self.assertTrue)

    def partition(source: Observable) -> List[Observable]:
        """The partially applied `partition` operator.

        Returns two observables which partition the observations of the
        source by the given function. The first will trigger
        observations for those values for which the predicate returns
        true. The second will trigger observations for those values
        where the predicate returns false. The predicate is executed
        once for each subscribed observer. Both also propagate all
        error observations arising from the source and each completes
        when the source completes.

        Args:
            source: Source obserable to partition.

        Returns:
            A list of observables. The first triggers when the
            predicate returns True, and the second triggers when the
            predicate returns False.
        """

        published = source.pipe(
            ops.publish(),
            ops.ref_count()
        )
        return [
            published.pipe(ops.filter(predicate)),
            published.pipe(ops.filter(lambda x: not predicate(x)))
        ]

        def create():
            def subscribe(o):
                o.on_next(1)
                o.on_next(2)
                return lambda: None

            return Observable.create(subscribe)

 async def go():
     nonlocal result
     source = Observable.throw(error)
     try:
         result = await source
     except Exception as ex:
         result = ex

            def on_next(inner_source: Observable):
                nonlocal source

                d = SingleAssignmentDisposable()
                with source.lock:
                    latest[0] += 1
                    _id = latest[0]
                has_latest[0] = True
                inner_subscription.disposable = d

                # Check if Future or Observable
                inner_source = from_future(inner_source) if is_future(inner_source) else inner_source

                def on_next(x: Any) -> None:
                    if latest[0] == _id:
                        observer.on_next(x)

                def on_error(e: Exception) -> None:
                    if latest[0] == _id:
                        observer.on_error(e)

                def on_completed() -> None:
                    if latest[0] == _id:
                        has_latest[0] = False
                        if is_stopped[0]:
                            observer.on_completed()

                d.disposable = inner_source.subscribe_(on_next, on_error, on_completed, scheduler=scheduler)

def time_interval(self, scheduler):
    """Records the time interval between consecutive values in an
    observable sequence.

    1 - res = source.time_interval();
    2 - res = source.time_interval(Scheduler.timeout)

    Keyword arguments:
    scheduler -- [Optional] Scheduler used to compute time intervals. If
        not specified, the timeout scheduler is used.

    Return An observable sequence with time interval information on values.
    """

    source = self
    scheduler = scheduler or timeout_scheduler

    def defer():
        last = [scheduler.now]

        def selector(x):
            now = scheduler.now
            span = now - last[0]
            last[0] = now
            return TimeInterval(value=x, interval=span)

        return source.map(selector)
    return Observable.defer(defer)

    def last(source: Observable) -> Observable:
        """Partially applied last operator.

        Returns the last element of an observable sequence that
        satisfies the condition in the predicate if specified, else
        the last element.

        Examples:
            >>> res = last(source)

        Args:
            source: Source observable to get last item from.

        Returns:
            An observable sequence containing the last element in the
            observable sequence that satisfies the condition in the
            predicate.
        """

        if predicate:
            return source.pipe(
                operators.filter(predicate),
                operators.last()
            )

        return last_or_default_async(source, False)