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

  def omega(self):
    return self

  def eval(self):
    return self.source

  def run(self, observer, cancel, setSink):
    sink = self.Sink(observer, cancel)
    setSink(sink)
    return self.source.subscribeSafe(sink)

  class Sink(rx.linq.sink.Sink):
    def __init__(self, observer, cancel):
      super(AsObservable.Sink, self).__init__(observer, cancel)

    def onNext(self, value):
      self.observer.onNext(value)

    def onError(self, exception):
      self.observer.onError(exception)
      self.dispose()

    def onCompleted(self):
      self.observer.onCompleted()
      self.dispose() class Buffer(Producer):
  def __init__(self, source, count=0, skip=0, timeSpan=0, timeShift=0, scheduler=None):
    if skip == 0:
      skip = count

    self.source = source
    self.count = count # length of each buffer
    self.skip = skip # number of elements to skip between creation of buffers
    self.timeShift = timeShift
    self.timeSpan = timeSpan
    self.scheduler = scheduler

  def run(self, observer, cancel, setSink):
    if self.scheduler == None:
      sink = self.SinkWithCount(self, observer, cancel)
      setSink(sink)
      return sink.run()
    elif self.count > 0:
      sink = self.SinkWithCountAndTimeSpan(self, observer, cancel)
      setSink(sink)
      return sink.run()
    else:
      if self.timeSpan == self.timeShift:
        sink = self.SinkWithTimeSpan(self, observer, cancel)
        setSink(sink)
        return sink.run()
      else:
        sink = self.SinkWithTimerAndTimeSpan(self, observer, cancel)
        setSink(sink)
        return sink.run()

  class SinkWithCount(rx.linq.sink.Sink):
    def __init__(self, parent, observer, cancel):
      super(Buffer.SinkWithCount, self).__init__(observer, cancel)
      self.parent = parent

    def run(self):
      self.queue = deque()
      self.n = 0

      self.createWindow()

      return self.parent.source.subscribeSafe(self)

    def createWindow(self):
      s = []
      self.queue.append(s)

    def onNext(self, value):
      for s in self.queue:
        s.append(value)

      c = self.n - self.parent.count + 1

      if c >= 0 and c % self.parent.skip == 0:
        s = self.queue.popleft()

        if len(s) > 0:
          self.observer.onNext(s)

      self.n += 1

      if self.n % self.parent.skip == 0:
        self.createWindow()

    def onError(self, exception):
      while len(self.queue) > 0:
        self.queue.popleft().clear()

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

    def onCompleted(self):
      while len(self.queue) > 0:
        s = self.queue.popleft()

        if len(s) > 0:
          self.observer.onNext(s)

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

  class SinkWithTimeSpan(rx.linq.sink.Sink):
    def __init__(self, parent, observer, cancel):
      super(Buffer.SinkWithTimeSpan, self).__init__(observer, cancel)
      self.parent = parent

    def run(self):
      self.gate = RLock()
      self.list = []

      d = self.parent.scheduler.schedulePeriodic(self.parent.timeSpan, self.tick)
      s = self.parent.source.subscribeSafe(self)

      return CompositeDisposable(d, s)

    def tick(self):
      with self.gate:
        self.observer.onNext(self.list)
        self.list = []

    def onNext(self, value):
      with self.gate:
        self.list.append(value)

    def onError(self, exception):
      with self.gate:
        self.list.clear()
        self.observer.onError(exception)
        self.dispose()

    def onCompleted(self):
      with self.gate:
        self.observer.onNext(self.list)
        self.observer.onCompleted()
        self.dispose()


  class SinkWithTimerAndTimeSpan(rx.linq.sink.Sink):
    def __init__(self, parent, observer, cancel):
      super(Buffer.SinkWithTimerAndTimeSpan, self).__init__(observer, cancel)
      self.parent = parent

    def run(self):
      self.totalTime = 0
      self.nextShift = self.parent.timeShift
      self.nextSpan = self.parent.timeSpan

      self.queue = deque()
      self.gate = RLock()

      self.timerDisposable = SerialDisposable()

      self.createWindow()
      self.createTimer()

      subscription = self.parent.source.subscribeSafe(self)

      return CompositeDisposable(self.timerDisposable, subscription)

    def createWindow(self):
      s = []
      self.queue.append(s)

    def createTimer(self):
      m = SingleAssignmentDisposable()
      self.timerDisposable.disposable = m

      isSpan = False
      isShift = False

      if self.nextSpan == self.nextShift:
        isSpan = True
        isShift = True
      elif self.nextShift < self.nextShift:
        isSpan = True
      else:
        isShift = True

      newTotalTime = self.nextSpan if isSpan else self.nextShift
      ts = newTotalTime - self.totalTime
      self.totalTime = newTotalTime

      if isSpan:
        self.nextSpan += self.parent.timeShift
      if isShift:
        self.nextShift += self.parent.timeShift

      m.disposable = self.parent.scheduler.scheduleWithRelativeAndState(
        Struct(isSpan=isSpan, isShift=isShift),
        ts,
        self.tick
      )

    def tick(self, scheduler, state):
      with self.gate:
        if state.isSpan:
          s = self.queue.popleft()
          self.observer.onNext(s)

        if state.isShift:
          self.createWindow()

      self.createTimer()

      return Disposable.empty()

    def onNext(self, value):
      with self.gate:
        for s in self.queue:
          s.append(value)

    def onError(self, exception):
      with self.gate:
        while len(self.queue) > 0:
          self.queue.popleft().clear()

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

    def onCompleted(self):
      with self.gate:
        while len(self.queue) > 0:
          s = self.queue.popleft()
          self.observer.onNext(s)

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

  class SinkWithCountAndTimeSpan(rx.linq.sink.Sink):
    def __init__(self, parent, observer, cancel):
      super(Buffer.SinkWithCountAndTimeSpan, self).__init__(observer, cancel)
      self.parent = parent

    def run(self):
      self.gate = RLock()
      self.list = []
      self.n = 0
      self.windowId = 0

      self.timerDisposable = SerialDisposable()
      self.createTimer(0)

      subscription = self.parent.source.subscribeSafe(self)

      return CompositeDisposable(self.timerDisposable, subscription)

    def createTimer(self, wId):
      m = SingleAssignmentDisposable()
      self.timerDisposable.disposable = m

      m.disposable = self.parent.scheduler.scheduleWithRelativeAndState(
        wId,
        self.parent.timeSpan,
        self.tick
      )

    def tick(self, scheduler, wId):
      d = Disposable.empty()

      newId = 0

      with self.gate:
        if wId != self.windowId:
          return d

        self.n = 0
        self.windowId += 1
        newId = self.windowId

        res = self.list
        self.list = []
        self.observer.onNext(res)

      self.createTimer(newId)

      return d

    def onNext(self, value):
      newWindow = False
      newId = 0

      with self.gate:
        self.list.append(value)
        self.n += 1

        if self.n == self.parent.count:
          newWindow = True
          self.windowId += 1
          newId = self.windowId

          res = self.list
          self.list = []
          self.n = 0
          self.observer.onNext(res)

      if newWindow:
        self.createTimer(newId)

    def onError(self, exception):
      with self.gate:
        self.list.clear()
        self.observer.onError(exception)
        self.dispose()

    def onCompleted(self):
      with self.gate:
        self.observer.onNext(self.list)
        self.observer.onCompleted()
        self.dispose() class Dematerialize(Producer):
  def __init__(self, source):
    self.source = source

  def run(self, observer, cancel, setSink):
    sink = self.Sink(observer, cancel)
    setSink(sink)
    return self.source.subscribeSafe(sink)

  class Sink(rx.linq.sink.Sink):
    def __init__(self, observer, cancel):
      super(Dematerialize.Sink, self).__init__(observer, cancel)

    def onNext(self, value):
      if value.kind == Notification.KIND_NEXT:
        self.observer.onNext(value.value)
      elif value.kind == Notification.KIND_ERROR:
        self.observer.onError(value.exception)
        self.dispose()
      elif value.kind == Notification.KIND_COMPLETED:
        self.observer.onCompleted()
        self.dispose()
      else:
        self.observer.onError(Exception("Unknown notification kind %s"%value.kind))

    def onError(self, exception):
      self.observer.onError(exception)
      self.dispose()

    def onCompleted(self):
      self.observer.onCompleted()
      self.dispose() class DistinctUntilChanged(Producer):
  def __init__(self, source, keySelector, equals):
    self.source = source
    self.keySelector = keySelector
    self.equals = equals

  def run(self, observer, cancel, setSink):
    sink = self.Sink(self, observer, cancel)
    setSink(sink)
    return self.source.subscribeSafe(sink)

  class Sink(rx.linq.sink.Sink):
    def __init__(self, parent, observer, cancel):
      super(DistinctUntilChanged.Sink, self).__init__(observer, cancel)
      self.parent = parent
      self.currentKey = None
      self.hasCurrentKey = False

    def onNext(self, value):
      try:
        key = self.parent.keySelector(value)
      except Exception as e:
        self.observer.onError(e)
        self.dispose()
      else:
        equal = False

        if self.hasCurrentKey:
          try:
            equal = self.parent.equals(self.currentKey, key)
          except Exception as e:
            self.observer.onError(e)
            self.dispose()

        if not self.hasCurrentKey or not equal:
          self.hasCurrentKey = True
          self.currentKey = key
          self.observer.onNext(value)

    def onError(self, exception):
      self.observer.onError(exception)
      self.dispose()

    def onCompleted(self):
      self.observer.onCompleted()
      self.dispose() class Do(Producer):
  def __init__(self, source, onNext, onError, onCompleted):
    self.source = source
    self.onNextAction = onNext
    self.onErrorAction = onError
    self.onCompletedAction = onCompleted

  def run(self, observer, cancel, setSink):
    sink = self.Sink(self, observer, cancel)
    setSink(sink)
    return self.source.subscribeSafe(sink)

  class Sink(rx.linq.sink.Sink):
    def __init__(self, parent, observer, cancel):
      super(Do.Sink, self).__init__(observer, cancel)
      self.parent = parent

    def onNext(self, value):
      try:
        self.parent.onNextAction(value)
      except Exception as e:
        self.observer.onError(e)
        self.dispose()
      else:
        self.observer.onNext(value)

    def onError(self, exception):
      try:
        self.parent.onErrorAction(exception)
      except Exception as e:
        self.observer.onError(e)
        self.dispose()
      else:
        self.observer.onNext(exception)
        self.dispose()

    def onCompleted(self):
      try:
        self.parent.onCompletedAction()
      except Exception as e:
        self.observer.onError(e)
        self.dispose()
      else:
        self.observer.onCompleted()
        self.dispose() class Finally(Producer):
  def __init__(self, source, action):
    self.source = source
    self.action = action

  def run(self, observer, cancel, setSink):
    sink = self.Sink(self, observer, cancel)
    setSink(sink)
    return sink.run()

  class Sink(rx.linq.sink.Sink):
    def __init__(self, parent, observer, cancel):
      super(Finally.Sink, self).__init__(observer, cancel)
      self.parent = parent

    def run(self):
      def dispose():
        try:
          subscription.dispose()
        finally:
          self.parent.action()

      subscription = self.parent.source.subscribeSafe(self)
      return Disposable.create(dispose)

    def onNext(self, value):
      self.observer.onNext(value)

    def onError(self, exception):
      self.observer.onError(exception)
      self.dispose()

    def onCompleted(self):
      self.observer.onCompleted()
      self.dispose() class IgnoreElements(Producer):
  def __init__(self, source):
    self.source = source

  def run(self, observer, cancel, setSink):
    sink = self.Sink(observer, cancel)
    setSink(sink)
    return self.source.subscribeSafe(sink)

  class Sink(rx.linq.sink.Sink):
    def __init__(self, observer, cancel):
      super(IgnoreElements.Sink, self).__init__(observer, cancel)

    def onNext(self, value):
      pass

    def onError(self, exception):
      self.observer.onError(exception)
      self.dispose()

    def onCompleted(self):
      self.observer.onCompleted()
      self.dispose() class Materialize(Producer):
  def __init__(self, source):
    self.source = source

  def dematerialize(self):
    return self.source.asObservable()

  def run(self, observer, cancel, setSink):
    sink = self.Sink(observer, cancel)
    setSink(sink)
    return self.source.subscribeSafe(sink)

  class Sink(rx.linq.sink.Sink):
    def __init__(self, observer, cancel):
      super(Materialize.Sink, self).__init__(observer, cancel)

    def onNext(self, value):
      self.observer.onNext(Notification.createOnNext(value))

    def onError(self, exception):
      self.observer.onNext(Notification.createOnError(exception))
      self.observer.onCompleted()
      self.dispose()

    def onCompleted(self):
      self.observer.onNext(Notification.createOnCompleted())
      self.observer.onCompleted()
      self.dispose() class ScanWithSeed(Producer):
  def __init__(self, source, seed, accumulator):
    self.source = source
    self.seed = seed
    self.accumulator = accumulator

  def run(self, observer, cancel, setSink):
    sink = self.Sink(self, observer, cancel)
    setSink(sink)
    return self.source.subscribeSafe(sink)

  class Sink(rx.linq.sink.Sink):
    def __init__(self, parent, observer, cancel):
      super(ScanWithSeed.Sink, self).__init__(observer, cancel)
      self.parent = parent
      self.accumulation = None
      self.hasAccumulation = False

    def onNext(self, value):
      try:
        if self.hasAccumulation:
          self.accumulation = self.parent.accumulator(self.accumulation, value)
        else:
          self.accumulation = self.parent.accumulator(self.parent.seed, value)
          self.hasAccumulation = True
      except Exception as e:
        self.observer.onError(e)
        self.dispose()
      else:
        self.observer.onNext(self.accumulation)

    def onError(self, exception):
      self.observer.onError(exception)
      self.dispose()

    def onCompleted(self):
      self.observer.onCompleted()
      self.dispose() class ScanWithoutSeed(Producer):
  def __init__(self, source, accumulator):
    self.source = source
    self.accumulator = accumulator

  def run(self, observer, cancel, setSink):
    sink = self.Sink(self, observer, cancel)
    setSink(sink)
    return self.source.subscribeSafe(sink)

  class Sink(rx.linq.sink.Sink):
    def __init__(self, parent, observer, cancel):
      super(ScanWithoutSeed.Sink, self).__init__(observer, cancel)
      self.parent = parent
      self.accumulation = None
      self.hasAccumulation = False

    def onNext(self, value):
      try:
        if self.hasAccumulation:
          self.accumulation = self.parent.accumulator(self.accumulation, value)
        else:
          self.accumulation = value
          self.hasAccumulation = True
      except Exception as e:
        self.observer.onError(e)
        self.dispose()
      else:
        self.observer.onNext(self.accumulation)

    def onError(self, exception):
      self.observer.onError(exception)
      self.dispose()

    def onCompleted(self):
      self.observer.onCompleted()
      self.dispose() class SkipLastCount(Producer):
  def __init__(self, source, count):
    self.source = source
    self.count = count

  def run(self, observer, cancel, setSink):
    sink = self.Sink(self, observer, cancel)
    setSink(sink)
    return self.source.subscribeSafe(sink)

  class Sink(rx.linq.sink.Sink):
    def __init__(self, parent, observer, cancel):
      super(SkipLastCount.Sink, self).__init__(observer, cancel)
      self.parent = parent
      self.queue = deque()

    def onNext(self, value):
      self.queue.append(value)

      if len(self.queue) > self.parent.count:
        self.observer.onNext(self.queue.popleft())

    def onError(self, exception):
      self.observer.onError(exception)
      self.dispose()

    def onCompleted(self):
      self.observer.onCompleted()
      self.dispose() class SkipLastTime(Producer):
  def __init__(self, source, duration, scheduler):
    self.source = source
    self.duration = duration
    self.scheduler = scheduler

  def run(self, observer, cancel, setSink):
    sink = self.Sink(self, observer, cancel)
    setSink(sink)
    return sink.run()

  class Sink(rx.linq.sink.Sink):
    def __init__(self, parent, observer, cancel):
      super(SkipLastTime.Sink, self).__init__(observer, cancel)
      self.parent = parent

    def run(self):
      self.startTime = self.parent.scheduler.now()

      return self.parent.subscribeSafe(self)

    def elapsed(self):
      return self.parent.scheduler.now() - self.startTime

    def onNext(self, value):
      now = self.elapsed()

      self.queue.append(Struct(value=value,timeStamp=now))

      while len(self.queue) > 0:
        current = self.queue.popleft()

        if now - current.timeStamp >= self.parent.duration:
          self.observer.onNext(current.value)
        else:
          self.queue.appendleft(current)
          break

    def onError(self, exception):
      self.observer.onError(exception)
      self.dispose()

    def onCompleted(self):
      now = self.elapsed()

      while len(self.queue) > 0:
        current = self.queue.popleft()

        if now - current.timeStamp >= self.parent.duration:
          self.observer.onNext(current.value)
        else:
          self.queue.appendleft(current)
          break

      self.observer.onCompleted()
      self.dispose() class TakeLastCount(Producer):
  def __init__(self, source, count, scheduler):
    self.source = source
    self.count = count
    self.scheduler = scheduler

  def run(self, observer, cancel, setSink):
    sink = self.Sink(self, observer, cancel)
    setSink(sink)
    return sink.run()

  class Sink(rx.linq.sink.Sink):
    def __init__(self, parent, observer, cancel):
      super(TakeLastCount.Sink, self).__init__(observer, cancel)
      self.parent = parent
      self.queue = deque()

    def run(self):
      self.subscription = SingleAssignmentDisposable()
      self.loopDisposable = SingleAssignmentDisposable()

      self.subscription.disposable = self.parent.source.subscribeSafe(self)

      return CompositeDisposable(self.subscription, self.loopDisposable)

    def onNext(self, value):
      self.queue.append(value)

      if len(self.queue) > self.parent.count:
        self.queue.popleft()

    def onError(self, exception):
      self.observer.onError(exception)
      self.dispose()

    def onCompleted(self):
      self.subscription.dispose()

      scheduler = self.parent.scheduler
      if scheduler.isLongRunning:
        self.loopDisposable.disposable = scheduler.scheduleLongRunning(self.loop)
      else:
        self.loopDisposable.disposable = scheduler.scheduleRecursive(self.loopRec)

    def loopRec(self, recurse):
      if len(self.queue) > 0:
        self.observer.onNext(self.queue.popleft())
        recurse()
      else:
        self.observer.onCompleted()
        self.dispose()

    def loop(self, cancel):
      while not cancel.isDisposed:
        if len(self.queue) == 0:
          self.observer.onCompleted()
          break
        else:
          self.observer.onNext(self.queue.popleft())

      self.dispose() class TakeLastTime(Producer):
  def __init__(self, source, duration, scheduler):
    self.source = source
    self.duration = duration
    self.scheduler = scheduler

  def run(self, observer, cancel, setSink):
    sink = self.Sink(self, observer, cancel)
    setSink(sink)
    return sink.run()

  class Sink(rx.linq.sink.Sink):
    def __init__(self, parent, observer, cancel):
      super(TakeLastTime.Sink, self).__init__(observer, cancel)
      self.parent = parent

    def run(self):
      self.subscription = SingleAssignmentDisposable()
      self.loop = SingleAssignmentDisposable()

      self.startTime = self.parent.scheduler.now()
      self.subscription.disposable = self.parent.source.subscribeSafe(self)

      return CompositeDisposable(self.subscription, self.loop)

    def elapsed(self):
      return self.parent.scheduler.now() - self.startTime

    def trim(self, now):
      while len(self.queue) > 0:
        current = self.queue.popleft()

        if current.interval < self.parent.duration:
          self.queue.appendleft(current)
          break

    def onNext(self, value):
      now = self.elapsed()

      self.queue.append(Struct(value=value,interval=now))
      self.trim(now)

    def onError(self, exception):
      self.observer.onError(exception)
      self.dispose()

    def onCompleted(self):
      self.subscription.dispose()

      now = self.elapsed()
      self.trim(now)

      scheduler = self.parent.scheduler
      if scheduler.isLongRunning:
        self.loop.disposable = scheduler.scheduleLongRunning(self.loop)
      else:
        self.loop.disposable = scheduler.scheduleRecursive(self.loopRec)

    def loopRec(self, recurse):
      if len(self.queue) > 0:
        self.observer.onNext(self.queue.popleft().value)
        recurse()
      else:
        self.observer.onCompleted()
        self.dispose()

    def loop(self, cancel):
      while not cancel.isDisposed:
        if len(self.queue) == 0:
          self.observer.onCompleted()
          break
        else:
          self.observer.onNext(self.queue.popleft().value)

      self.dispose() class TakeLastBufferCount(Producer):
  def __init__(self, source, count):
    self.source = source
    self.count = count

  def run(self, observer, cancel, setSink):
    sink = self.Sink(self, observer, cancel)
    setSink(sink)
    return sink.run()

  class Sink(rx.linq.sink.Sink):
    def __init__(self, parent, observer, cancel):
      super(TakeLastBufferCount.Sink, self).__init__(observer, cancel)
      self.parent = parent
      self.queue = deque()

    def run(self):
      self.subscription = SingleAssignmentDisposable()
      self.loop = SingleAssignmentDisposable()

      self.subscription.disposable = self.parent.source.subscribeSafe(self)

      return CompositeDisposable(self.subscription, self.loop)

    def onNext(self, value):
      self.queue.append(value)

      if len(self.queue) > self.parent.count:
        self.queue.popleft()

    def onError(self, exception):
      self.observer.onError(exception)
      self.dispose()

    def onCompleted(self):
      res = list(self.queue)

      self.observer.onNext(res)
      self.observer.onCompleted()
      self.dispose() class TakeLastBufferTime(Producer):
  def __init__(self, source, duration, scheduler):
    self.source = source
    self.duration = duration
    self.scheduler = scheduler

  def run(self, observer, cancel, setSink):
    sink = self.Sink(self, observer, cancel)
    setSink(sink)
    return sink.run()

  class Sink(rx.linq.sink.Sink):
    def __init__(self, parent, observer, cancel):
      super(TakeLastBufferTime.Sink, self).__init__(observer, cancel)
      self.parent = parent

    def run(self):
      self.subscription = SingleAssignmentDisposable()
      self.loop = SingleAssignmentDisposable()

      self.startTime = self.parent.scheduler.now()
      self.subscription.disposable = self.parent.source.subscribeSafe(self)

      return CompositeDisposable(self.subscription, self.loop)

    def elapsed(self):
      return self.parent.scheduler.now() - self.startTime

    def trim(self, now):
      while len(self.queue) > 0:
        current = self.queue.popleft()

        if current.interval < self.parent.duration:
          self.queue.appendleft(current)
          break

    def onNext(self, value):
      now = self.elapsed()

      self.queue.append(Struct(value=value,interval=now))
      self.trim(now)

    def onError(self, exception):
      self.observer.onError(exception)
      self.dispose()

    def onCompleted(self):
      self.subscription.dispose()

      now = self.elapsed()
      self.trim(now)

      res = list([x.value for x in self.queue])

      self.observer.onNext(res)
      self.observer.onCompleted()
      self.dispose() class Window(Producer):
  def __init__(self, source, count=0, skip=0, timeSpan=0, timeShift=0, scheduler=None):
    if skip == 0:
      skip = count

    self.source = source
    self.count = count # length of each window
    self.skip = skip # number of elements to skip between creation of windows
    self.timeShift = timeShift
    self.timeSpan = timeSpan
    self.scheduler = scheduler

  def run(self, observer, cancel, setSink):
    if self.scheduler == None:
      sink = self.SinkWithCount(self, observer, cancel)
      setSink(sink)
      return sink.run()
    elif self.count > 0:
      sink = self.SinkWithCountAndTimeSpan(self, observer, cancel)
      setSink(sink)
      return sink.run()
    else:
      if self.timeSpan == self.timeShift:
        sink = self.SinkWithTimeSpan(self, observer, cancel)
        setSink(sink)
        return sink.run()
      else:
        sink = self.SinkWithTimerAndTimeSpan(self, observer, cancel)
        setSink(sink)
        return sink.run()

  class SinkWithCount(rx.linq.sink.Sink):
    def __init__(self, parent, observer, cancel):
      super(Window.SinkWithCount, self).__init__(observer, cancel)
      self.parent = parent

    def run(self):
      self.queue = deque()
      self.n = 0
      self.m = SingleAssignmentDisposable()
      self.refCountDisposable = RefCountDisposable(self.m)

      firstWindow = self.createWindow()
      self.observer.onNext(firstWindow)

      self.m.disposable = self.parent.source.subscribeSafe(self)

      return self.refCountDisposable

    def createWindow(self):
      s = Subject()
      self.queue.append(s)
      # AddRef was originally WindowObservable but this is just an alias for AddRef
      return AddRef(s, self.refCountDisposable)

    def onNext(self, value):
      for s in self.queue:
        s.onNext(value)

      c = self.n - self.parent.count + 1

      if c >= 0 and c % self.parent.skip == 0:
        s = self.queue.popleft()
        s.onCompleted()

      self.n += 1

      if self.n % self.parent.skip == 0:
        newWindow = self.createWindow()
        self.observer.onNext(newWindow)

    def onError(self, exception):
      while len(self.queue) > 0:
        self.queue.popleft.onError(exception)

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

    def onCompleted(self):
      while len(self.queue) > 0:
        self.queue.popleft().onCompleted()

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

  class SinkWithTimeSpan(rx.linq.sink.Sink):
    def __init__(self, parent, observer, cancel):
      super(Window.SinkWithTimeSpan, self).__init__(observer, cancel)
      self.parent = parent

    def run(self):
      self.gate = RLock()

      groupDisposable = CompositeDisposable()
      self.refCountDisposable = RefCountDisposable(groupDisposable)

      self.createWindow()

      groupDisposable.add(self.parent.scheduler.schedulePeriodic(self.parent.timeSpan, self.tick))
      groupDisposable.add(self.parent.source.subscribeSafe(self))

      return self.refCountDisposable

    def tick(self):
      with self.gate:
        self.subject.onCompleted()
        self.createWindow()

    def createWindow(self):
      self.subject = Subject()
      self.observer.onNext(AddRef(self.subject, self.refCountDisposable))

    def onNext(self, value):
      with self.gate:
        self.list.append(value)

    def onError(self, exception):
      with self.gate:
        self.subject.onError(exception)

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

    def onCompleted(self):
      with self.gate:
        self.subject.onCompleted()

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


  class SinkWithTimerAndTimeSpan(rx.linq.sink.Sink):
    def __init__(self, parent, observer, cancel):
      super(Window.SinkWithTimerAndTimeSpan, self).__init__(observer, cancel)
      self.parent = parent

    def run(self):
      self.totalTime = 0
      self.nextShift = self.parent.timeShift
      self.nextSpan = self.parent.timeSpan

      self.gate = RLock()
      self.queue = deque()

      self.timerDisposable = SerialDisposable()

      groupDisposable = CompositeDisposable(self.timerDisposable)
      self.refCountDisposable = RefCountDisposable(groupDisposable)

      self.createWindow()
      self.createTimer()

      groupDisposable.add(self.parent.source.subscribeSafe(self))

      return self.refCountDisposable

    def createWindow(self):
      s = Subject()
      self.queue.append(s)
      self.observer.onNext(AddRef(s, self.refCountDisposable))

    def createTimer(self):
      m = SingleAssignmentDisposable()
      self.timerDisposable.disposable = m

      isSpan = False
      isShift = False

      if self.nextSpan == self.nextShift:
        isSpan = True
        isShift = True
      elif self.nextShift < self.nextShift:
        isSpan = True
      else:
        isShift = True

      newTotalTime = self.nextSpan if isSpan else self.nextShift
      ts = newTotalTime - self.totalTime
      self.totalTime = newTotalTime

      if isSpan:
        self.nextSpan += self.parent.timeShift
      if isShift:
        self.nextShift += self.parent.timeShift

      m.disposable = self.parent.scheduler.scheduleWithRelativeAndState(
        Struct(isSpan=isSpan, isShift=isShift),
        ts,
        self.tick
      )

    def tick(self, scheduler, state):
      with self.gate:
        if state.isSpan:
          s = self.queue.popleft()
          s.onCompleted()

        if state.isShift:
          self.createWindow()

      self.createTimer()

      return Disposable.empty()

    def onNext(self, value):
      with self.gate:
        for s in self.queue:
          s.onNext(value)

    def onError(self, exception):
      with self.gate:
        for o in self.queue:
          o.onError(exception)

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

    def onCompleted(self):
      with self.gate:
        for o in self.queue:
          o.onCompleted()

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

  class SinkWithCountAndTimeSpan(rx.linq.sink.Sink):
    def __init__(self, parent, observer, cancel):
      super(Window.SinkWithCountAndTimeSpan, self).__init__(observer, cancel)
      self.parent = parent

    def run(self):
      self.gate = RLock()
      self.s = Subject()
      self.n = 0
      self.windowId = 0

      self.timerDisposable = SerialDisposable()
      groupDisposable = CompositeDisposable(self.timerDisposable)
      self.refCountDisposable = RefCountDisposable(groupDisposable)

      # AddRef was originally WindowObservable but this is just an alias for AddRef
      self.observer.onNext(AddRef(self.s, self.refCountDisposable))
      self.createTimer(0)

      groupDisposable.add(self.parent.source.subscribeSafe(self))

      return self.refCountDisposable

    def createTimer(self, wId):
      m = SingleAssignmentDisposable()
      self.timerDisposable.disposable = m

      m.disposable = self.parent.scheduler.scheduleWithRelativeAndState(
        wId,
        self.parent.timeSpan,
        self.tick
      )

    def tick(self, scheduler, wId):
      d = Disposable.empty()

      newId = 0

      with self.gate:
        if wId != self.windowId:
          return d

        self.n = 0
        self.windowId += 1
        newId = self.windowId

        self.s.onCompleted()
        self.s = Subject()
        self.observer.onNext(AddRef(self.s, self.refCountDisposable))

      self.createTimer(newId)

      return d

    def onNext(self, value):
      newWindow = False
      newId = 0

      with self.gate:
        self.s.onNext(value)
        self.n += 1

        if self.n == self.parent.count:
          newWindow = True
          self.n = 0
          self.windowId += 1
          newId = self.windowId

          self.s.onCompleted()
          self.s = Subject()
          self.observer.onNext(AddRef(self.s, self.refCountDisposable))

      if newWindow:
        self.createTimer(newId)

    def onError(self, exception):
      with self.gate:
        self.s.onError(exception)
        self.observer.onError(exception)
        self.dispose()

    def onCompleted(self):
      with self.gate:
        self.s.onCompleted()
        self.observer.onCompleted()
        self.dispose() def defaultEquals(x, y): return x == y def identity(x): return x def noop(*args, **kwargs): pass 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 Scheduler(object, metaclass=MetaScheduler):
  """Provides a set of static properties to access commonly
  used Schedulers and implements all the scheduling methods that
  then use the overrides of the implementation."""

  @staticmethod
  def invokeAction(scheduler, action):
    action()
    return Disposable.empty()

  @staticmethod
  def invokeRecImmediate(scheduler, pair):
    state = pair[0]
    action = pair[1]

    scheduled = RecursiveScheduledFunction(action, scheduler)
    scheduled.run(state)

    return scheduled.group

  @staticmethod
  def invokeRecDate(scheduler, pair, method):
    state = pair[0]
    action = pair[1]

    scheduled = RecursiveScheduledFunction(action, scheduler, method)
    scheduled.run(state)

    return scheduled.group

  @staticmethod
  def now():
    return defaultNow()

  @staticmethod
  def normalize(timeSpan):
    if timeSpan < 0:
      return 0
    else:
      return timeSpan

  def catchException(self, handler):
    return CatchScheduler(self, handler)

  # longrunning scheduling
  # action takes as parameter: state and cancel
  # and returns None
  def scheduleLongRunning(self, action):
    return self.scheduleLongRunningWithState(None, lambda s, cancel: action(cancel))

  ## commented out such that one can test if a scheduler can schedule long running
  # def scheduleLongRunningWithState(self, state, action):
  #   raise NotImplementedError()
  @property
  def isLongRunning(self):
    return hasattr(self, "scheduleLongRunningWithState")

  # periodic scheduling
  # action takes as parameter: state
  # and returns: state
  def schedulePeriodic(self, period, action):
    return self.schedulePeriodicWithState(None, period, lambda s: action())

  def schedulePeriodicWithState(self, state, period, action):
    def gated():
      state = action(state)

    timer = PeriodicTimer(period, gated)

    return timer.start()

  # once scheduling
  # action takes as parameter: scheduler, state
  # and returns: disposable to cancel
  def schedule(self, action):
    return self._scheduleCore(action, Scheduler.invokeAction)

  def scheduleWithState(self, state, action):
    return self._scheduleCore(state, action)

  def scheduleWithRelative(self, dueTime, action):
    return self._scheduleRelativeCore(action, dueTime, Scheduler.invokeAction)

  def scheduleWithRelativeAndState(self, state, dueTime, action):
    return self._scheduleRelativeCore(state, dueTime, action)

  def scheduleWithAbsolute(self, dueTime, action):
    return self._scheduleAbsoluteCore(action, dueTime, Scheduler.invokeAction)

  def scheduleWithAbsoluteAndState(self, state, dueTime, action):
    return self._scheduleAbsoluteCore(state, dueTime, action)

  # recursive scheduling
  # action takes as parameter: continuation([state], [period])
  # and returns: None
  def scheduleRecursive(self, action):
    return self.scheduleRecursiveWithState(
      None,
      lambda _, _continuation: action(lambda: _continuation(None))
    )

  def scheduleRecursiveWithState(self, state, action):
    return self.scheduleWithState((state, action), Scheduler.invokeRecImmediate)

  def scheduleRecursiveWithRelative(self, dueTime, action):
    return self.scheduleRecursiveWithRelativeAndState(
      None,
      dueTime,
      lambda _, _continuation: action(lambda dt: _continuation(None, dt))
    )

  def scheduleRecursiveWithRelativeAndState(self, state, dueTime, action):
    return self._scheduleRelativeCore(
      (state, action),
      dueTime,
      lambda s, p: Scheduler.invokeRecDate(s, p, 'scheduleWithRelativeAndState')
    )

  def scheduleRecursiveWithAbsolute(self, dueTime, action):
    return self.scheduleRecursiveWithAbsoluteAndState(
      action,
      lambda _action, _self: _action(lambda: _self(_action))
    )

  def scheduleRecursiveWithAbsoluteAndState(self, state, dueTime, action):
    return self._scheduleAbsoluteCore(
      (state, action),
      dueTime,
      lambda s, p: Scheduler.invokeRecDate(s, p, 'scheduleWithAbsoluteAndState')
    )

  def _scheduleCore(self, state, action):
    raise NotImplementedError()

  def _scheduleRelativeCore(self, state, dueTime, action):
    raise NotImplementedError()

  def _scheduleAbsoluteCore(self, state, dueTime, action):
    raise NotImplementedError()
from .asObservable import AsObservable
from .buffer import Buffer
from .dematerialize import Dematerialize
from .distinctUntilChanged import DistinctUntilChanged
from .do import Do
from .finallyOp import Finally
from .ignoreElements import IgnoreElements
from .materialize import Materialize
from .scan import ScanWithSeed, ScanWithoutSeed
from .skipLast import SkipLastCount, SkipLastTime
from .takeLast import TakeLastCount, TakeLastTime
from .takeLastBuffer import TakeLastBufferCount, TakeLastBufferTime
from .window import Window
from rx.internal import defaultEquals, identity, noop
from rx.observable import Observable
from rx.scheduler import Scheduler
import itertools
  assert callable(equals)

  return DistinctUntilChanged(self, keySelector, equals)
Observable.distinctUntilChanged = distinctUntilChanged

def do(self, onNext=noop, onError=noop, onCompleted=noop):
  assert isinstance(self, Observable)
  assert callable(onNext)
  assert callable(onError)
  assert callable(onCompleted)

  return Do(self, onNext, onError, onCompleted)
Observable.do = do

def doFinally(self, action):
  assert isinstance(self, Observable)
  assert callable(action)

  return Finally(self, action)
Observable.doFinally = doFinally

def ignoreElements(self):
  assert isinstance(self, Observable)

  if isinstance(self, IgnoreElements):
    return self.omega()
  else:
    return IgnoreElements(self)
Observable.ignoreElements = ignoreElements

def materialize(self):
  #
  # NOTE: Peephole optimization of xs.Dematerialize().Materialize() should not be performed. It's possible for xs to
  #       contain multiple terminal notifications, which won't survive a Dematerialize().Materialize() chain. In case
  #       a reduction to xs.AsObservable() would be performed, those notification elements would survive.
  #
  assert isinstance(self, Observable)

  return Materialize(self)
Observable.materialize = materialize

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))
Observable.repeatSelf = repeatSelf

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))
Observable.retry = retry

def scan(self, accumulator, seed=None):
  assert isinstance(self, Observable)
  assert callable(accumulator)

  if seed == None:
    return ScanWithoutSeed(self, accumulator)
  else:
    return ScanWithSeed(self, seed, accumulator)
Observable.scan = scan

def skipLast(self, count):
  assert isinstance(self, Observable)

  return SkipLastCount(self, count)
Observable.skipLast = skipLast

def skipLastWithTime(self, time, scheduler=Scheduler.timeBasedOperation):
  assert isinstance(self, Observable)
  assert isinstance(scheduler, Scheduler)

  return SkipLastTime(self, time, scheduler)
Observable.skipLastWithTime = skipLastWithTime

def startWith(self, *values):
  assert isinstance(self, Observable)

  return Observable.fromIterable(list(values)).concat(self)
Observable.startWith = startWith

def takeLast(self, count, scheduler=Scheduler.timeBasedOperation):
  assert isinstance(self, Observable)

  return TakeLastCount(self, count, scheduler)
Observable.takeLast = takeLast

def takeLastWithTime(self, time, scheduler=Scheduler.timeBasedOperation):
  assert isinstance(self, Observable)
  assert isinstance(scheduler, Scheduler)

  return TakeLastTime(self, time, scheduler)
Observable.takeLastWithTime = takeLastWithTime

def takeLastBuffer(self, count):
  assert isinstance(self, Observable)

  return TakeLastBufferCount(self, count)
Observable.takeLastBuffer = takeLastBuffer

def takeLastBufferWithTime(self, time, scheduler=Scheduler.timeBasedOperation):
  assert isinstance(self, Observable)
  assert isinstance(scheduler, Scheduler)

  return TakeLastBufferTime(self, time, scheduler)
Observable.takeLastBufferWithTime = takeLastBufferWithTime

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

  if skip == None:
    return Window(self, count=count, skip=count)
  else:

    return Window(self, count=count, skip=skip)