# Path: rx/observable.py
# class Producer(Observable):
#   """Base class for implementation of query operators, providing
#   performance benefits over the use of Observable.Create"""
# 
#   def subscribeCore(self, observer):
#     return self.subscribeRaw(observer, True)
# 
#   def subscribeRaw(self, observer, enableSafequard):
#     sink = SingleAssignmentDisposable()
#     subscription = SingleAssignmentDisposable()
# 
#     d = CompositeDisposable(sink, subscription)
# 
#     if enableSafequard:
#       observer = AutoDetachObserver(observer, d)
# 
#     def assignSink(s):
#       sink.disposable = s
# 
#     def scheduled():
#       subscription.disposable = self.run(observer, subscription, assignSink)
#       return Disposable.empty()
# 
#     if Scheduler.currentThread.isScheduleRequired():
#       Scheduler.currentThread.schedule(scheduled)
#     else:
#       scheduled()
# 
#     return d
# 
#   def run(self, observer, cancel, setSink):
#     raise NotImplementedError()


from rx.observable import Producer
import rx.linq.sink
      while n > 0 and not cancel.isDisposed:
        self.observer.onNext(value)
        n -= 1

      if not cancel.isDisposed:
        self.observer.onCompleted()

      self.dispose()

    def loopInf(self, cancel):
      value = self.parent.value

      while not cancel.isDisposed:
        self.observer.onNext(value)

      self.dispose()

    def loopRec(self, n, recurse):
      if n > 0:
        self.observer.onNext(self.parent.value)
        n -= 1

      if n == 0:
        self.observer.onCompleted()
        self.dispose()
        return

      recurse(n)

    def loopRecInf(self, recurse):

# Path: rx/observable.py
# class Producer(Observable):
#   """Base class for implementation of query operators, providing
#   performance benefits over the use of Observable.Create"""
# 
#   def subscribeCore(self, observer):
#     return self.subscribeRaw(observer, True)
# 
#   def subscribeRaw(self, observer, enableSafequard):
#     sink = SingleAssignmentDisposable()
#     subscription = SingleAssignmentDisposable()
# 
#     d = CompositeDisposable(sink, subscription)
# 
#     if enableSafequard:
#       observer = AutoDetachObserver(observer, d)
# 
#     def assignSink(s):
#       sink.disposable = s
# 
#     def scheduled():
#       subscription.disposable = self.run(observer, subscription, assignSink)
#       return Disposable.empty()
# 
#     if Scheduler.currentThread.isScheduleRequired():
#       Scheduler.currentThread.schedule(scheduled)
#     else:
#       scheduled()
# 
#     return d
# 
#   def run(self, observer, cancel, setSink):
#     raise NotImplementedError()


# Path: rx/linq/repeat.py
from rx.observable import Producer
import rx.linq.sink

      while n > 0 and not cancel.isDisposed:
        self.observer.onNext(value)
        n -= 1

      if not cancel.isDisposed:
        self.observer.onCompleted()

      self.dispose()

    def loopInf(self, cancel):
      value = self.parent.value

      while not cancel.isDisposed:
        self.observer.onNext(value)

      self.dispose()

    def loopRec(self, n, recurse):
      if n > 0:
        self.observer.onNext(self.parent.value)
        n -= 1

      if n == 0:
        self.observer.onCompleted()
        self.dispose()
        return

      recurse(n)

    def loopRecInf(self, recurse):

      self.observer.onNext(self.parent.value)