# 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/observer.py
# class ObserveOnObserver(ScheduledObserver):
#   def __init__(self, scheduler, observer, cancel):
#     super(ObserveOnObserver, self).__init__(scheduler, observer)
#     self.cancel = Atomic(cancel, self.lock)
# 
#   def onNextCore(self, value):
#     super(ObserveOnObserver, self).onNextCore(value)
#     self.ensureActive()
# 
#   def onErrorCore(self, exception):
#     super(ObserveOnObserver, self).onErrorCore(exception)
#     self.ensureActive()
# 
#   def onCompletedCore(self):
#     super(ObserveOnObserver, self).onCompletedCore()
#     self.ensureActive()
# 
#   def dispose(self):
#     super(ObserveOnObserver, self).dispose()
# 
#     old = self.cancel.exchange(None)
# 
#     if old != None:
#       old.dispose()


from rx.observable import Producer
from rx.observer import ObserveOnObserver


class ObserveOn(Producer):
  def __init__(self, source, scheduler):
    self.source = source
    self.scheduler = scheduler

  def run(self, observer, cancel, setSink):

# 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/observer.py
# class ObserveOnObserver(ScheduledObserver):
#   def __init__(self, scheduler, observer, cancel):
#     super(ObserveOnObserver, self).__init__(scheduler, observer)
#     self.cancel = Atomic(cancel, self.lock)
# 
#   def onNextCore(self, value):
#     super(ObserveOnObserver, self).onNextCore(value)
#     self.ensureActive()
# 
#   def onErrorCore(self, exception):
#     super(ObserveOnObserver, self).onErrorCore(exception)
#     self.ensureActive()
# 
#   def onCompletedCore(self):
#     super(ObserveOnObserver, self).onCompletedCore()
#     self.ensureActive()
# 
#   def dispose(self):
#     super(ObserveOnObserver, self).dispose()
# 
#     old = self.cancel.exchange(None)
# 
#     if old != None:
#       old.dispose()


# Path: rx/linq/observeOn.py
from rx.observable import Producer
from rx.observer import ObserveOnObserver



class ObserveOn(Producer):
  def __init__(self, source, scheduler):
    self.source = source
    self.scheduler = scheduler

  def run(self, observer, cancel, setSink):

    sink = ObserveOnObserver(self.scheduler, observer, cancel)