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 class Disposable(object):
  """Represents a disposable object"""

  def dispose(self):
    pass

  def __enter__(self):
    return self

  def __exit__(self, exc_type, exc_value, traceback):
    self.dispose()

  @staticmethod
  def create(action):
    return AnonymouseDisposable(action)

  @staticmethod
  def empty():
    return disposableEmpty def errorIfDisposed(disposable):
  if disposable.isDisposed:
    raise Exception("Object has been disposed") class Struct(object):
  def __init__(self, **entries):
    self.__dict__.update(entries)

  def __eq__(self, other):
    if not isinstance(other, Struct):
      return False

    return self.__dict__ == other.__dict__

  def __repr__(self):
    return repr(self.__dict__) class Observable(object):
  """Provides all extension methods to Observable"""

  def subscribe(self, observerOrOnNext=noop, onError=noop, onComplete=noop):
    observer = observerOrOnNext

    if observerOrOnNext == None or callable(observerOrOnNext):
      observer = Observer.create(observerOrOnNext, onError, onComplete)

    return self.subscribeCore(observer)

  def subscribeCore(self, observer):
    raise NotImplementedError()

  def subscribeSafe(self, observer):
    if isinstance(self, ObservableBase):
      return self.subscribeCore(observer)
    elif isinstance(self, Producer):
      return self.subscribeRaw(observer, False)

    d = Disposable.empty()

    try:
      d = self.subscribeCore(observer)
    except Exception as e:
      observer.onError(e)

    return d class DisposedObserver(Observer):
  def onNext(self, value):
    raise Exception("Object has been disposed")

  def onError(self, exception):
    raise Exception("Object has been disposed")

  def onCompleted(self):
    raise Exception("Object has been disposed") class DoneObserver(Observer):
  def __init__(self, exception=None):
    super(DoneObserver, self).__init__()
    self.exception = exception

  def onNext(self, value):
    pass

  def onError(self, exception):
    pass

  def onCompleted(self):
    pass class NoopObserver(Observer):
  def onNext(self, value):
    pass

  def onError(self, exception):
    pass

  def onCompleted(self):
    pass class ListObserver(Observer):
  def __init__(self, observers):
    super(ListObserver, self).__init__()
    self.observers = observers

  def onNext(self, value):
    for observer in self.observers:
      observer.onNext(value)

  def onError(self, exception):
    for observer in self.observers:
      observer.onError(exception)

  def onCompleted(self):
    for observer in self.observers:
      observer.onCompleted()

  def add(self, observer):
    return ListObserver(self.observers + (observer,))

  def remove(self, observer):
    if observer not in self.observers:
      return self

    index = self.observers.index(observer)
    newObservers = self.observers[0:index] + self.observers[index+1:]

    return ListObserver(newObservers) 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() class ScheduledObserver(ObserverBase):
  STOPPED = 0
  RUNNING = 1
  PENDING = 2
  FAULTED = 9

  def __init__(self, scheduler, observer):
    super(ScheduledObserver, self).__init__()
    self.scheduler = scheduler
    self.observer = observer
    self.state = Atomic(ScheduledObserver.STOPPED, self.lock)
    self.disposable = SerialDisposable()

    self.failed = False
    self.exception = None
    self.completed = False

    self.queue = Queue()
    self.dispatcherJob = None
    self.dispatcherEvent = Semaphore(0)

  def ensureDispatcher(self):
    if self.dispatcherJob != None:
      return

    with self.lock:
      if self.dispatcherJob == None:
        self.dispatcherJob = self.scheduler.scheduleLongRunning(self.dispatch)
        self.disposable.disposable = CompositeDisposable(
          self.dispatcherJob,
          Disposable.create(self.dispatcherEvent.release)
        )

  def dispatch(self, cancel):
    while True:
      self.dispatcherEvent.acquire()

      if cancel.isDisposed:
        return

      while True:
        next = self.queue.get()

        try:
          self.observer.onNext(next)
        except Exception as e:
          self.clearQueue()
          raise e

        self.dispatcherEvent.acquire()

        if cancel.isDisposed:
          return

      if self.failed:
        self.observer.onError(self.exception)
        self.dispose()

        return

      if self.completed:
        self.observer.onCompleted()
        self.dispose()

        return

  def ensureActive(self, n = 1):
    if self.scheduler.isLongRunning:
      while n > 0:
        self.dispatcherEvent.release()
        n -= 1

        self.ensureDispatcher()
    else:
      self.ensureActiveSlow()

  def ensureActiveSlow(self):
    isOwner = False

    while True:
      old = self.state.compareExchange(
        ScheduledObserver.RUNNING,
        ScheduledObserver.STOPPED
      )

      if old == ScheduledObserver.STOPPED:
        isOwner = True
        break
      elif old == ScheduledObserver.FAULTED:
        return
      elif (
          (old == ScheduledObserver.PENDING or old == ScheduledObserver.RUNNING) and
          self.state.compareExchange(ScheduledObserver.PENDING, ScheduledObserver.RUNNING) == ScheduledObserver.RUNNING
        ):
        break

    if isOwner:
      self.disposable = self.scheduler.scheduleRecursiveWithState(None, self.run)

  def run(self, state, continuation):
    next = None

    while True:
      try:
        next = self.queue.get_nowait()
      except Empty:
        next = None

      if next != None:
        break

      if self.failed:
        # wait until the queue is drained
        if not self.queue.empty():
          continue

        self.state.value = ScheduledObserver.STOPPED

        self.observer.onError(self.exception)
        self.dispose()

        return

      if self.completed:
        # wait until the queue is drained
        if not self.queue.empty():
          continue

        self.state.value = ScheduledObserver.STOPPED

        self.observer.onCompleted()
        self.dispose()

        return

      old = self.state.compareExchange(
        ScheduledObserver.STOPPED,
        ScheduledObserver.RUNNING
      )

      if old == ScheduledObserver.RUNNING or old == ScheduledObserver.FAULTED:
        return

      # assert(old == ScheduledObserver.PENDING)

      self.state.value = ScheduledObserver.RUNNING
    #end while

    # we found an item, so next != None
    self.state.value = ScheduledObserver.RUNNING

    try:
      self.observer.onNext(next)
    except Exception as e:
      self.state.value = ScheduledObserver.FAULTED
      self.clearQueue()

      raise e

    continuation(state)

  def onNextCore(self, value):
    self.queue.put(value)

  def onErrorCore(self, exception):
    self.exception = exception
    self.failed = True

  def onCompletedCore(self):
    self.completed = True

  def clearQueue(self):
    try:
      while True:
        self.queue.get()
    except Empty:
      pass

  def dispose(self):
    super(ScheduledObserver, self).dispose()
    self.disposable.dispose() class MetaScheduler(type):
class Scheduler(object, metaclass=MetaScheduler):
class CatchWrapper(object):
class CatchScheduler(Scheduler):
class CurrentThreadScheduler(Scheduler):
class DefaultScheduler(Scheduler):
class VirtualTimeScheduler(Scheduler):
class HistoricalScheduler(VirtualTimeScheduler):
class ImmediateScheduler(Scheduler):
  class AsyncLockScheduler(Scheduler):
class RecursiveScheduledFunction(object):
class PeriodicTimer(object):
class ScheduledItem(object):
  def currentThread(cls):
  def immediate(cls):
  def default(cls):
  def constantTimeOperations(cls):
  def tailRecursion(cls):
  def iteration(cls):
  def timeBasedOperation(cls):
  def asyncConversions(cls):
  def invokeAction(scheduler, action):
  def invokeRecImmediate(scheduler, pair):
  def invokeRecDate(scheduler, pair, method):
  def now():
  def normalize(timeSpan):
  def catchException(self, handler):
  def scheduleLongRunning(self, action):
  def isLongRunning(self):
  def schedulePeriodic(self, period, action):
  def schedulePeriodicWithState(self, state, period, action):
    def gated():
  def schedule(self, action):
  def scheduleWithState(self, state, action):
  def scheduleWithRelative(self, dueTime, action):
  def scheduleWithRelativeAndState(self, state, dueTime, action):
  def scheduleWithAbsolute(self, dueTime, action):
  def scheduleWithAbsoluteAndState(self, state, dueTime, action):
  def scheduleRecursive(self, action):
  def scheduleRecursiveWithState(self, state, action):
  def scheduleRecursiveWithRelative(self, dueTime, action):
  def scheduleRecursiveWithRelativeAndState(self, state, dueTime, action):
  def scheduleRecursiveWithAbsolute(self, dueTime, action):
  def scheduleRecursiveWithAbsoluteAndState(self, state, dueTime, action):
  def _scheduleCore(self, state, action):
  def _scheduleRelativeCore(self, state, dueTime, action):
  def _scheduleAbsoluteCore(self, state, dueTime, action):
  def __init__(self, parent, action):
  def __call__(self, _self, _state):
  def __init__(self, scheduler, handler):
  def now(self):
  def _scheduleCore(self, state, action):
  def _scheduleRelativeCore(self, state, dueTime, action):
  def _scheduleAbsoluteCore(self, state, dueTime, action):
  def _clone(self, scheduler):
  def _wrap(self, action):
  def _getRecursiveWrapper(self, scheduler):
  def schedulePeriodicWithState(self, state, period, action):
    def scheduled(_state):
  def __init__(self):
  def isScheduleRequired(self):
  def ensureTrampoline(self, action):
  def _queue():
    def fget(self):
    def fset(self, value):
    def fdel(self):
  def _init(self):
  def _dispose(self):
  def _run(self):
  def _scheduleCore(self, state, action):
  def _scheduleRelativeCore(self, state, dueTime, action):
  def _scheduleAbsoluteCore(self, state, dueTime, action):
  def __init__(self):
  def _scheduleCore(self, state, action):
    def scheduled():
  def _scheduleRelativeCore(self, state, dueTime, action):
    def scheduled():
  def _scheduleAbsoluteCore(self, state, dueTime, action):
  def schedulePeriodicWithState(self, state, interval, action):
    def gated():
  def scheduleLongRunningWithState(self, state, action):
    def run():
  def __init__(self, clock, comparer):
  def now(self):
  def _scheduleCore(self, state, action):
  def _scheduleRelativeCore(self, state, dueTime, action):
  def _scheduleAbsoluteCore(self, state, dueTime, action):
  def schedulePeriodicWithState(self, state, period, action):
    def _replace(disposable):
  def scheduleRelativeWithState(self, state, dueTime, action):
  def scheduleRelative(self, dueTime, action):
  def scheduleAbsolute(self, dueTime, action):
  def scheduleAbsoluteWithState(self, state, dueTime, action):
  def start(self, until=None):
  def stop(self):
  def advanceTo(self, time):
  def advanceBy(self, time):
  def sleep(self, time):
  def getNext(self):
  def __init__(self, initialClock = 0, comparer = defaultSubComparer):
  def add(self, absolute, relative):
  def toDateTimeOffset(self, absolute):
  def toRelative(self, timeSpan):
  def __init__(self):
  def _scheduleCore(self, state, action):
  def _scheduleRelativeCore(self, state, dueTime, action):
  def _scheduleAbsoluteCore(self, state, dueTime, action):
    def __init__(self):
    def _scheduleCore(self, state, action):
      def gated():
    def _scheduleRelativeCore(self, state, dueTime, action):
      def gated():
    def _scheduleAbsoluteCore(self, state, dueTime, action):
  def __init__(self, action, scheduler, method = None):
  def run(self, state):
  def actionCallback(self, newState, dueTime = None):
  def schedulerCallback(self, scheduler, state):
  def __init__(self, interval, action):
  def start(self):
  def cancel(self):
  def _execute(self):
  def __init__(self, scheduler, state, action, dueTime, comparer = defaultSubComparer):
  def invoke(self):
  def isCancelled(self):
  def invokeCore(self):
  def compareTo(self, other):
  def __lt__(self, other):
from rx.concurrency import Atomic
from rx.disposable import Disposable
from rx.internal import errorIfDisposed, Struct
from rx.observable import Observable
from rx.observer import DisposedObserver, DoneObserver, NoopObserver, ListObserver, Observer, ScheduledObserver
from rx.scheduler import currentThreadScheduler
from threading import RLock
import sys
      ex = self.exception

    if ex != None:
      observer.onError(ex)
    else:
      observer.onCompleted()

    return Disposable.empty()

  def unsubscribe(self, observer):
    with self.gate:
      if observer in self.observers:
        self.observers.remove(observer)

  def dispose(self):
    with self.gate:
      self.isDisposed = True
      self.observers = []
      self.value = None
      self.exception = None


class ReplaySubject(Observable, Observer):
  def __init__(self, bufferSize = sys.maxsize, window = sys.maxsize, scheduler = currentThreadScheduler):
    super(ReplaySubject, self).__init__()
    self.bufferSize = bufferSize
    self.window = window
    self.scheduler = scheduler
    self.q = []
    self.observers = []
    self.isStopped = False
    self.isDisposed = False
    self.hasError = False
    self.exception = None
    self.gate = RLock()

  @property
  def hasObservers(self):
    os = self.observers
    return os != None and len(os) > 0

  def _trim(self, now):
    self.q = self.q[-self.bufferSize:]
    self.q = [node for node in self.q if self.window >= (now - node.interval)]

  def onCompleted(self):
    os = []

    with self.gate:
      errorIfDisposed(self)

      if not self.isStopped:
        os = list(self.observers)
        now = self.scheduler.now()

        self.isStopped = True
        self.observers = []

        self._trim(now)

        for observer in os:
          observer.onCompleted()

    for observer in os:
      observer.ensureActive()

  def onError(self, exception):
    os = []

    with self.gate:
      errorIfDisposed(self)

      if not self.isStopped:
        os = list(self.observers)
        now = self.scheduler.now()

        self.isStopped = True
        self.observers = []
        self.exception = exception

        self._trim(now)

        for observer in os:
          observer.onError(exception)

    for observer in os:
      observer.ensureActive()

  def onNext(self, value):
    os = []

    with self.gate:
      errorIfDisposed(self)

      if not self.isStopped:
        os = list(self.observers)
        now = self.scheduler.now()

        self.q.append(Struct(interval = now, value = value))

        for observer in os:
          observer.onNext(value)

    for observer in os:
      observer.ensureActive()

  class Subscription(Disposable):
    def __init__(self, subject, observer):
      self.subject = subject
      self.observer = Atomic(observer)

    def dispose(self):
      old = self.observer.exchange(None)

      if old != None:
        old.dispose()
        self.subject.unsubscribe(old)
        self.subject = None

  def subscribeCore(self, observer):

    so = ScheduledObserver(self.scheduler, observer)