def test_select_with_index_throws(self):
     try:
         return Observable.return_value(1) \
             .select(lambda x, index: x) \
             .subscribe(lambda x: _raise('ex'))
     except RxException:
         pass
 
     try:
         return Observable.throw_exception('ex') \
             .select(lambda x, index: x) \
             .subscribe(lambda x: x, lambda ex: _raise(ex))
     except RxException:
         pass
 
     try:
         return Observable.empty() \
             .select(lambda x, index: x) \
             .subscribe(lambda x: x, lambda ex: _, lambda : _raise('ex'))
     except RxException:
         pass
 
     try:
         return Observable.create(lambda o: _raise('ex')) \
             .select(lambda x, index: x) \
             .subscribe()
     except RxException:
         pass

    def test_repeat_observable_repeat_count_throws(self):
        scheduler1 = TestScheduler()
        xs = Observable.return_value(1, scheduler1).repeat(3)
        xs.subscribe(lambda x: _raise('ex'))
        
        try:
            return scheduler1.start()
        except RxException:
            pass

        scheduler2 = TestScheduler()
        ys = Observable.throwException('ex1', scheduler2).repeat(3)
        ys.subscribe(lambda ex: _raise('ex2'))
        
        try:
            return scheduler2.start()
        except RxException:
            pass

        scheduler3 = TestScheduler()
        zs = Observable.return_value(1, scheduler3).repeat(100)
        d = zs.subscribe(on_complete=lambda: _raise('ex3'))
        
        scheduler3.schedule_absolute(10, lambda: d.dispose())
        
        scheduler3.start()
        xss = Observable.create(lambda o: _raise('ex4')).repeat(3)
        try:
            return xss.subscribe()
        except RxException:
            pass

    def test_retry_observable_throws(self):
        scheduler1 = TestScheduler()
        xs = Observable.return_value(1, scheduler1).retry()
        xs.subscribe(lambda x: _raise('ex'))
        
        try:
            return scheduler1.start()
        except RxException:
            pass

        scheduler2 = TestScheduler()
        ys = Observable.throw_exception('ex', scheduler2).retry()
        d = ys.subscribe(on_error=lambda ex: _raise('ex'))
        
        scheduler2.schedule_absolute(210, lambda: d.dispose())
        
        scheduler2.start()
        scheduler3 = TestScheduler()
        zs = Observable.return_value(1, scheduler3).retry()
        zs.subscribe(on_completed=lambda: _raise('ex'))
        
        try:
            return scheduler3.start()
        except RxException:
            pass

        xss = Observable.create(lambda o: _raise('ex')).retry()
        try:
            return xss.subscribe()
        except RxException:
            pass

def retry(self, count = None):
  assert isinstance(self, Observable)

  if count == None:
    return Observable.catchFallback(itertools.repeat(self))
  else:
    return Observable.catchFallback(itertools.repeat(self, count))

def repeatSelf(self, count = None):
  assert isinstance(self, Observable)

  if count == None:
    return Observable.concat(itertools.repeat(self))
  else:
    return Observable.concat(itertools.repeat(self, count))

    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 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 _start(self, app_context, **kwargs):
     if self.__input is None:
         self.__input = app_context.inst_data_mgr.get_series(self.input.name)
     self.__input.subject.subscribe(on_next=self.on_update)
     if self.__output_bar_type == BarType.Time:
         current_ts = self.__clock.now()
         next_ts = Bar.get_next_bar_start_time(current_ts, self.__output_size)
         diff = next_ts - current_ts
         Observable.timer(int(diff), self.__output_size * 1000, self.__clock.scheduler).subscribe(
             on_next=self.publish)

    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 branch(condition, thenSource, schedulerOrElseSource=None):
  assert callable(condition)
  assert isinstance(thenSource, Observable)

  if schedulerOrElseSource == None:
    return If(condition, thenSource, Observable.empty())
  elif isinstance(schedulerOrElseSource, Scheduler):
    return If(condition, thenSource, Observable.empty(schedulerOrElseSource))
  else:
    assert isinstance(schedulerOrElseSource, Observable)

    return If(condition, thenSource, schedulerOrElseSource)

def case(selector, sources, schedulerOrDefaultSource=None):
  assert callable(selector)
  assert isinstance(sources, dict)

  if schedulerOrDefaultSource == None:
    return Case(selector, sources, Observable.empty())
  elif isinstance(schedulerOrDefaultSource, Scheduler):
    return Case(selector, sources, Observable.empty(schedulerOrDefaultSource))
  else:
    assert isinstance(schedulerOrDefaultSource, Observable)

    return Case(selector, sources, schedulerOrDefaultSource)

def timeoutIndividual(self, durationSelector, firstTimeout=None, other=None):
  assert isinstance(self, Observable)

  if firstTimeout == None:
    firstTimeout = Observable.never()
  if other == None:
    other = Observable.throw(TimeoutException())

  assert isinstance(firstTimeout, Observable)
  assert isinstance(other, Observable)

  return TimeoutObservable(self, firstTimeout, durationSelector, other)

    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 while_do(cls, condition, source):
      """Repeats source as long as condition holds emulating a while loop.
      
      Keyword arguments:
      condition -- {Function} The condition which determines if the source 
          will be repeated.
      source -- {Observable} The observable sequence that will be run if the 
          condition function returns true.
 
      Returns an observable {Observable} sequence which is repeated as long 
      as the condition holds."""
      
      source = Observable.from_future(source)
      return Observable.concat(Enumerable.while_do(condition, source))    

 def test_select_throws(self):
     try:
         Observable.return_value(1) \
             .select(lambda x, y: x) \
             .subscribe(lambda x: _raise("ex"))
     except RxException:
         pass
 
     try:
         Observable.throw_exception('ex') \
             .select(lambda x, y: x) \
             .subscribe(on_error=lambda ex: _raise(ex))
     except RxException:
         pass
 
     try:
         Observable.empty() \
             .select(lambda x, y: x) \
             .subscribe(lambda x: x, lambda ex: ex, lambda: _raise('ex'))
     except RxException:
         pass
 
     def subscribe(observer):
         _raise('ex')
 
     try:
         Observable.create(subscribe) \
             .select(lambda x: x) \
             .subscribe()
     except RxException:
         pass

    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.select(selector)
        return Observable.defer(defer)

 def test_empty_observer_throw_exception(self):
     scheduler = TestScheduler()
     xs = Observable.empty(scheduler)
     xs.subscribe(lambda x: None, lambda ex: None, lambda: _raise('ex'))
     
     with self.assertRaises(RxException):
         scheduler.start()

def while_do(cls, condition, source):
    """Repeats source as long as condition holds emulating a while loop.

    Keyword arguments:
    :param types.FunctionType condition: The condition which determines if the
        source will be repeated.
    :param Observable source: The observable sequence that will be run if the
        condition function returns true.

    :returns: An observable sequence which is repeated as long as the condition
        holds.
    :rtype: Observable
    """

    source = Observable.from_future(source)
    return Observable.concat(Enumerable.while_do(condition, source))

    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 args[0].now:
            scheduler = args[0]
            sources = args[1:]
        else:
            scheduler = immediate_scheduler
            sources = args[0]

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

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

def repeat(cls, value=None, repeat_count=None, scheduler=None):
    """Generates an observable sequence that repeats the given element the
    specified number of times, using the specified scheduler to send out
    observer messages.

    1 - res = rx.Observable.repeat(42)
    2 - res = rx.Observable.repeat(42, 4)
    3 - res = rx.Observable.repeat(42, 4, Rx.Scheduler.timeout)
    4 - res = rx.Observable.repeat(42, None, Rx.Scheduler.timeout)

    Keyword arguments:
    value -- Element to repeat.
    repeat_count -- [Optional] Number of times to repeat the element. If not
        specified, repeats indefinitely.
    scheduler -- Scheduler to run the producer loop on. If not specified,
        defaults to ImmediateScheduler.

    Returns an observable sequence that repeats the given element the
    specified number of times."""

    scheduler = scheduler or current_thread_scheduler
    if repeat_count == -1:
        repeat_count = None

    xs = Observable.return_value(value, scheduler)
    return xs.repeat(repeat_count)