# 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 Observer(Disposable):
#   """Represents the IObserver Interface.
#   Has some static helper methods attached"""
# 
#   @staticmethod
#   def create(onNext=None, onError=None, onCompleted=None):
#     if onNext == None:
#       onNext = noop
#     if onError == None:
#       onError = defaultError
#     if onCompleted == None:
#       onCompleted = noop
# 
#     return AnonymousObserver(onNext, onError, onCompleted)
# 
#   @staticmethod
#   def synchronize(observer, lock=None):
#     if lock == None:
#       lock = RLock()
# 
#     return SynchronizedObserver(observer, lock)
# 
#   @staticmethod
#   def fromNotifier(handler):
#     return AnonymousObserver(
#       lambda x: handler(Notification.createOnNext(x)),
#       lambda ex: handler(Notification.createOnError(ex)),
#       lambda: handler(Notification.createOnCompleted())
#     )
# 
#   def toNotifier(self):
#     return lambda n: n.accept(self)
# 
#   def asObserver(self):
#     return AnonymousObserver(self.onNext, self.onError, self.onCompleted)
# 
#   def checked(self):
#     return CheckedObserver(self)
# 
#   def notifyOn(self, scheduler):
#     return ScheduledObserver(scheduler, self)
# 
#   def onNext(self, value):
#     raise NotImplementedError()
# 
#   def onError(self, exception):
#     raise NotImplementedError()
# 
#   def onCompleted(self):
#     raise NotImplementedError()
# 
# Path: rx/concurrency.py
# class Atomic:
#   def __init__(self, value=None, lock=RLock()):
#     self.lock = lock
#     self._value = value
# 
#   def value():
#       doc = "The value property."
#       def fget(self):
#           return self._value
#       def fset(self, value):
#           self.exchange(value)
#       return locals()
#   value = property(**value())
# 
#   def exchange(self, value):
#     with self.lock:
#       old = self._value
#       self._value = value
#       return old
# 
#   def compareExchange(self, value, expected):
#     with self.lock:
#       old = self._value
# 
#       if old == expected:
#         self._value = value
# 
#       return old
# 
#   def inc(self, by=1):
#     with self.lock:
#       self._value += by
#       return self.value
# 
#   def dec(self, by=1):
#     with self.lock:
#       self._value -= by
#       return self.value


from rx.observable import Producer
from rx.observer import Observer
from rx.concurrency import Atomic


class ForEach(Producer):
  class Sink(Observer):
    def __init__(self, onNext, done):
      self.onNextAction = onNext
      self.doneAction = done

      self.stopped = Atomic(False)
      self.exception = None

    def onNext(self, value):
      if not self.stopped.value:
        try:
          self.onNextAction(value)
        except Exception as e:
          self.onError(e)

    def onError(self, exception):
      if not self.stopped.exchange(True):
        self.exception = exception
        self.doneAction()

    def onCompleted(self):
      if not self.stopped.exchange(True):
        self.doneAction()


# 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 Observer(Disposable):
#   """Represents the IObserver Interface.
#   Has some static helper methods attached"""
# 
#   @staticmethod
#   def create(onNext=None, onError=None, onCompleted=None):
#     if onNext == None:
#       onNext = noop
#     if onError == None:
#       onError = defaultError
#     if onCompleted == None:
#       onCompleted = noop
# 
#     return AnonymousObserver(onNext, onError, onCompleted)
# 
#   @staticmethod
#   def synchronize(observer, lock=None):
#     if lock == None:
#       lock = RLock()
# 
#     return SynchronizedObserver(observer, lock)
# 
#   @staticmethod
#   def fromNotifier(handler):
#     return AnonymousObserver(
#       lambda x: handler(Notification.createOnNext(x)),
#       lambda ex: handler(Notification.createOnError(ex)),
#       lambda: handler(Notification.createOnCompleted())
#     )
# 
#   def toNotifier(self):
#     return lambda n: n.accept(self)
# 
#   def asObserver(self):
#     return AnonymousObserver(self.onNext, self.onError, self.onCompleted)
# 
#   def checked(self):
#     return CheckedObserver(self)
# 
#   def notifyOn(self, scheduler):
#     return ScheduledObserver(scheduler, self)
# 
#   def onNext(self, value):
#     raise NotImplementedError()
# 
#   def onError(self, exception):
#     raise NotImplementedError()
# 
#   def onCompleted(self):
#     raise NotImplementedError()
# 
# Path: rx/concurrency.py
# class Atomic:
#   def __init__(self, value=None, lock=RLock()):
#     self.lock = lock
#     self._value = value
# 
#   def value():
#       doc = "The value property."
#       def fget(self):
#           return self._value
#       def fset(self, value):
#           self.exchange(value)
#       return locals()
#   value = property(**value())
# 
#   def exchange(self, value):
#     with self.lock:
#       old = self._value
#       self._value = value
#       return old
# 
#   def compareExchange(self, value, expected):
#     with self.lock:
#       old = self._value
# 
#       if old == expected:
#         self._value = value
# 
#       return old
# 
#   def inc(self, by=1):
#     with self.lock:
#       self._value += by
#       return self.value
# 
#   def dec(self, by=1):
#     with self.lock:
#       self._value -= by
#       return self.value


# Path: rx/linq/forEach.py
from rx.observable import Producer
from rx.observer import Observer
from rx.concurrency import Atomic



class ForEach(Producer):
  class Sink(Observer):
    def __init__(self, onNext, done):
      self.onNextAction = onNext
      self.doneAction = done

      self.stopped = Atomic(False)
      self.exception = None

    def onNext(self, value):
      if not self.stopped.value:
        try:
          self.onNextAction(value)
        except Exception as e:
          self.onError(e)

    def onError(self, exception):
      if not self.stopped.exchange(True):
        self.exception = exception
        self.doneAction()

    def onCompleted(self):
      if not self.stopped.exchange(True):
        self.doneAction()


  class EnumeratingSink(Observer):