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

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

  def getSources(self):
    return self.sources

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

    def onNext(self, value):
      self.found = True
      self.observer.onNext(value)

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

    def onCompleted(self):
      if not self.found:
        self.observer.onNext(self.parent.defaultValue)

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

  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(Distinct.Sink, self).__init__(observer, cancel)
      self.parent = parent
      self.hashSet = set()

    def onNext(self, value):
      try:
        key = self.parent.keySelector(value)
      except Exception as e:
        self.observer.onError(e)
        self.dispose()
      else:
        if key in self.hashSet:
          return
        else:
          self.hashSet.add(key)
          self.observer.onNext(value)

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

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

  def run(self, observer, cancel, setSink):
    self.groupDisposable = CompositeDisposable()
    self.refCountDisposable = RefCountDisposable(self.groupDisposable)

    sink = self.Sink(self, observer, cancel)
    setSink(sink)
    self.groupDisposable.add(self.source.subscribeSafe(sink))

    return self.refCountDisposable


  class Sink(rx.linq.sink.Sink):
    def __init__(self, parent, observer, cancel):
      super(GroupBy.Sink, self).__init__(observer, cancel)
      self.parent = parent
      self.map = {}
      self.null = None

    def onNext(self, value):
      key = None

      try:
        key = self.parent.keySelector(value)
      except Exception as e:
        self.onError(e)
        return

      fireNewMapEntry = False
      writer = None

      try:
        if key == None:
          if self.null == None:
            self.null = Subject()
            fireNewMapEntry = True

          writer = self.null
        else:
          if key in self.map:
            writer = self.map[key]
          else:
            writer = Subject()
            self.map[key] = writer
            fireNewMapEntry = True
      except Exception as e:
        self.onError(e)
        return

      if fireNewMapEntry:
        group = GroupObservable(key, writer, self.parent.refCountDisposable)
        self.observer.onNext(group)

      element = None

      try:
        element = self.parent.elementSelector(value)
      except Exception as e:
        self.onError(e)
      else:
        writer.onNext(element)

    def onError(self, exception):
      if self.null != None:
        self.null.onError(exception)

      for x in self.map.values():
        x.onError(exception)

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

    def onCompleted(self):
      if self.null != None:
        self.null.onCompleted()

      for x in self.map.values():
        x.onCompleted()

      self.observer.onCompleted()
      self.dispose() class GroupByUntil(Producer):
  def __init__(self, source, keySelector, elementSelector, durationSelector):
    self.source = source
    self.keySelector = keySelector
    self.elementSelector = elementSelector
    self.durationSelector = durationSelector

  def run(self, observer, cancel, setSink):
    self.groupDisposable = CompositeDisposable()
    self.refCountDisposable = RefCountDisposable(self.groupDisposable)

    sink = self.Sink(self, observer, cancel)
    setSink(sink)
    self.groupDisposable.add(self.source.subscribeSafe(sink))

    return self.refCountDisposable


  class Sink(rx.linq.sink.Sink):
    def __init__(self, parent, observer, cancel):
      super(GroupByUntil.Sink, self).__init__(observer, cancel)
      self.parent = parent
      self.map = {}
      self.null = None
      self.nullGate = RLock()
      self.nullGateForDeltas = RLock()
      self.observerGate = RLock()
      self.writerGate = RLock()

    def onNext(self, value):
      key = None

      try:
        key = self.parent.keySelector(value)
      except Exception as e:
        self.onError(e)
        return

      fireNewMapEntry = False
      writer = None

      try:
        if key == None:
          with self.nullGate:
            if self.null == None:
              self.null = Subject()
              fireNewMapEntry = True

            writer = self.null
        else:
          if key in self.map:
            writer = self.map[key]
          else:
            writer = Subject()
            self.map[key] = writer
            fireNewMapEntry = True
      except Exception as e:
        self.onError(e)
        return

      if fireNewMapEntry:
        group = GroupObservable(key, writer, self.parent.refCountDisposable)

        duration = None
        durationGroup = GroupObservable(key, writer)

        try:
          duration = self.parent.durationSelector(durationGroup)
        except Exception as e:
          self.onError(e)
          return

        with self.observerGate:
          self.observer.onNext(group)

        md = SingleAssignmentDisposable()
        self.parent.groupDisposable.add(md)
        md.disposable = duration.subscribeSafe(self.Delta(self, key, writer, md))

      element = None

      try:
        element = self.parent.elementSelector(value)
      except Exception as e:
        self.onError(e)
      else:
        with self.writerGate:
          writer.onNext(element)

    def onError(self, exception):
      #
      # NOTE: A race with OnCompleted triggered by a duration selector is fine when
      #       using Subject<T>. It will transition into a terminal state, making one
      #       of the two calls a no-op by swapping in a DoneObserver<T>.
      #
      null = None

      with self.nullGate:
        null = self.null

      if null != None:
        null.onError(exception)

      for x in self.map.values():
        x.onError(exception)

      with self.observerGate:
        self.observer.onError(exception)

      self.dispose()

    def onCompleted(self):
      #
      # NOTE: A race with OnCompleted triggered by a duration selector is fine when
      #       using Subject<T>. It will transition into a terminal state, making one
      #       of the two calls a no-op by swapping in a DoneObserver<T>.
      #
      null = None

      with self.nullGate:
        null = self.null

      if null != None:
        null.onCompleted()

      for x in self.map.values():
        x.onCompleted()

      with self.observerGate:
        self.observer.onCompleted()

      self.dispose()


    class Delta(Observer):
      def __init__(self, parent, key, writer, cancelSelf):
        self.parent = parent
        self.key = key
        self.writer = writer
        self.cancelSelf = cancelSelf

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

      def onError(self, exception):
        self.parent.onError(exception)
        self.cancelSelf.dispose()

      def onCompleted(self):
        if self.key == None:
          null = None

          with self.parent.nullGate:
            null = self.parent.null
            self.parent.null = None

          with self.parent.nullGateForDeltas:
            null.onCompleted()
        else:
          try:
            del self.parent.map[self.key]
          except KeyError:
            pass
          else:
            with self.parent.writerGate:
              self.writer.onCompleted()

        self.parent.parent.groupDisposable.remove(self.cancelSelf)
    # end Delta
  #end Sink class GroupJoin(Producer):
  def __init__(self, left, right, leftDurationSelector, rightDurationSelector, resultSelector):
    self.left = left
    self.right = right
    self.leftDurationSelector = leftDurationSelector
    self.rightDurationSelector = rightDurationSelector
    self.resultSelector = resultSelector

  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(GroupJoin.Sink, self).__init__(observer, cancel)
      self.parent = parent

    def run(self):
      self.gate = RLock()
      self.group = CompositeDisposable()
      self.refCount = RefCountDisposable(self.group)

      leftSubscription = SingleAssignmentDisposable()
      self.group.add(leftSubscription)
      self.leftID = 0
      self.leftMap = {}

      rightSubscription = SingleAssignmentDisposable()
      self.group.add(rightSubscription)
      self.rightID = 0
      self.rightMap = {}

      leftSubscription.disposable = self.parent.left.subscribeSafe(self.Left(self, leftSubscription))
      rightSubscription.disposable = self.parent.right.subscribeSafe(self.Right(self, rightSubscription))

      return self.refCount

    class Left(Observer):
      def __init__(self, parent, subscription):
        self.parent = parent
        self.subscription = subscription

      def expire(self, resourceId, group, resource):
        with self.parent.gate:
          if resourceId in self.parent.leftMap:
            del self.parent.leftMap[resourceId]
            group.onCompleted()

        self.parent.group.remove(resource)

      def onNext(self, value):
        s = Subject()
        resourceId = 0

        with self.parent.gate:
          self.parent.leftID += 1
          resourceId = self.parent.leftID
          self.parent.leftMap[resourceId] = s

        # AddRef was originally WindowObservable but this is just an alias for AddRef
        window = AddRef(s, self.parent.refCount)
        md = SingleAssignmentDisposable()
        self.parent.group.add(md)

        try:
          duration = self.parent.parent.leftDurationSelector(value)
        except Exception as e:
          self.onError(e)
          return
        else:
          md.disposable = duration.subscribeSafe(self.Delta(self, resourceId, s, md))

        try:
          result = self.parent.parent.resultSelector(value, window)
        except Exception as e:
          self.onError(e)
          return
        else:
          with self.parent.gate:
            self.parent.observer.onNext(result)

            for rightValue in self.parent.rightMap.values():
              s.onNext(rightValue)

      def onError(self, exception):
        with self.parent.gate:
          for o in self.parent.leftMap.values():
            o.onError(exception)

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

      def onCompleted(self):
        with self.parent.gate:
          self.parent.observer.onCompleted()
          self.parent.dispose()

        self.dispose()


      class Delta(Observer):
        """Expires parent on Next or Completed"""

        def __init__(self, parent, resourceId, group, resource):
          self.parent = parent
          self.resourceId = resourceId
          self.group = group
          self.resource = resource

        def onNext(self, value):
          self.parent.expire(self.resourceId, self.group, self.resource)

        def onError(self, exception):
          self.parent.onError(exception)

        def onCompleted(self):
          self.parent.expire(self.resourceId, self.group, self.resource)
      #end Delta
    #end Left

    class Right(Observer):
      def __init__(self, parent, subscription):
        self.parent = parent
        self.subscription = subscription


      def expire(self, resourceId, resource):
        with self.parent.gate:
          self.parent.rightMap.pop(resourceId, None)

        self.parent.group.remove(resource)

      def onNext(self, value):
        resourceId = 0

        with self.parent.gate:
          self.parent.rightID += 1
          resourceId = self.parent.rightID
          self.parent.rightMap[resourceId] = value

        md = SingleAssignmentDisposable()
        self.parent.group.add(md)

        try:
          duration = self.parent.parent.rightDurationSelector(value)
        except Exception as e:
          self.onError(e)
          return
        else:
          md.disposable = duration.subscribeSafe(self.Delta(self, resourceId, md))

        with self.parent.gate:
          for o in self.parent.leftMap.values():
            o.onNext(value)

      def onError(self, exception):
        with self.parent.gate:
          for o in self.parent.leftMap.values():
            o.onError(exception)

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

      def onCompleted(self):
        self.dispose()


      class Delta(Observer):
        """Expires parent on Next or Completed"""
        def __init__(self, parent, resourceId, resource):
          self.parent = parent
          self.resourceId = resourceId
          self.resource = resource

        def onNext(self, value):
          self.parent.expire(self.resourceId, self.resource)

        def onError(self, exception):
          self.parent.onError(exception)

        def onCompleted(self):
          self.parent.expire(self.resourceId, self.resource)
      #end Delta
    #end Right
  #end Sink class Join(Producer):
  def __init__(self, left, right, leftDurationSelector, rightDurationSelector, resultSelector):
    super(Join, self).__init__()
    self.left = left
    self.right = right
    self.leftDurationSelector = leftDurationSelector
    self.rightDurationSelector = rightDurationSelector
    self.resultSelector = resultSelector


  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(Join.Sink, self).__init__(observer, cancel)
      self.parent = parent

    def run(self):
      self.gate = RLock()
      self.group = CompositeDisposable()
      self.refCount = RefCountDisposable(self.group)

      leftSubscription = SingleAssignmentDisposable()
      self.group.add(leftSubscription)
      self.leftDone = False
      self.leftID = 0
      self.leftMap = {}

      rightSubscription = SingleAssignmentDisposable()
      self.group.add(rightSubscription)
      self.rightDone = False
      self.rightID = 0
      self.rightMap = {}

      leftSubscription.disposable = self.parent.left.subscribeSafe(self.Lambda(self, leftSubscription))
      rightSubscription.disposable = self.parent.right.subscribeSafe(self.Roh(self, rightSubscription))

      return self.refCount

    class Lambda(Observer):
      def __init__(self, parent, subscription):
        self.parent = parent
        self.subscription = subscription

      def expire(self, resourceId, resource):
        with self.parent.gate:
          if resourceId in self.parent.leftMap:
            del self.parent.leftMap[resourceId]

            if len(self.parent.leftMap) == 0 and self.parent.leftDone:
              self.parent.observer.onCompleted()
              self.parent.dispose()

        self.parent.group.remove(resource)

      def onNext(self, value):
        resourceId = 0

        with self.parent.gate:
          self.parent.leftID += 1
          resourceId = self.parent.leftID
          self.parent.leftMap[resourceId] = value

        # AddRef was originally WindowObservable but this is just an alias for AddRef
        md = SingleAssignmentDisposable()
        self.parent.group.add(md)

        try:
          duration = self.parent.parent.leftDurationSelector(value)
        except Exception as e:
          self.parent.observer.onError(e)
          self.parent.dispose()
          return
        else:
          md.disposable = duration.subscribeSafe(self.Delta(self, resourceId, md))

        with self.parent.gate:
          for rightValue in self.parent.rightMap.values():
            try:
              result = self.parent.parent.resultSelector(value, rightValue)
            except Exception as e:
              self.parent.observer.onError(e)
              self.parent.dispose()
              return
            else:
              self.parent.observer.onNext(result)

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

      def onCompleted(self):
        with self.parent.gate:
          self.parent.leftDone = True

          if self.parent.rightDone or len(self.parent.leftMap) == 0:
            self.parent.observer.onCompleted()
            self.parent.dispose()
          else:
            self.dispose()


      class Delta(Observer):
        """Expires parent on Next or Completed"""

        def __init__(self, parent, resourceId, resource):
          self.parent = parent
          self.resourceId = resourceId
          self.resource = resource

        def onNext(self, value):
          self.parent.expire(self.resourceId, self.resource)

        def onError(self, exception):
          self.parent.onError(exception)

        def onCompleted(self):
          self.parent.expire(self.resourceId, self.resource)
      #end Delta
    #end Lambda

    class Roh(Observer):
      def __init__(self, parent, subscription):
        self.parent = parent
        self.subscription = subscription

      def expire(self, resourceId, resource):
        with self.parent.gate:
          if resourceId in self.parent.rightMap:
            del self.parent.rightMap[resourceId]

            if len(self.parent.rightMap) == 0 and self.parent.rightDone:
              self.parent.observer.onCompleted()
              self.parent.dispose()

        self.parent.group.remove(resource)

      def onNext(self, value):
        resourceId = 0

        with self.parent.gate:
          self.parent.rightID += 1
          resourceId = self.parent.rightID
          self.parent.rightMap[resourceId] = value

        # AddRef was originally WindowObservable but this is just an alias for AddRef
        md = SingleAssignmentDisposable()
        self.parent.group.add(md)

        try:
          duration = self.parent.parent.rightDurationSelector(value)
        except Exception as e:
          self.parent.observer.onError(e)
          self.parent.dispose()
          return
        else:
          md.disposable = duration.subscribeSafe(self.Delta(self, resourceId, md))

        with self.parent.gate:
          for leftValue in self.parent.leftMap.values():
            try:
              result = self.parent.parent.resultSelector(leftValue, value)
            except Exception as e:
              self.parent.observer.onError(e)
              self.parent.dispose()
              return
            else:
              self.parent.observer.onNext(result)

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

      def onCompleted(self):
        with self.parent.gate:
          self.parent.rightDone = True

          if self.parent.leftDone or len(self.parent.rightMap) == 0:
            self.parent.observer.onCompleted()
            self.parent.dispose()
          else:
            self.dispose()


      class Delta(Observer):
        """Expires parent on Next or Completed"""

        def __init__(self, parent, resourceId, resource):
          self.parent = parent
          self.resourceId = resourceId
          self.resource = resource

        def onNext(self, value):
          self.parent.expire(self.resourceId, self.resource)

        def onError(self, exception):
          self.parent.onError(exception)

        def onCompleted(self):
          self.parent.expire(self.resourceId, self.resource)
    #end Roh
  #end Sink class OfType(Producer):
  def __init__(self, source, tpe):
    self.source = source
    self.tpe = tpe

  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(OfType.Sink, self).__init__(observer, cancel)
      self.parent = parent
      self.currentKey = None
      self.hasCurrentKey = False

    def onNext(self, value):
      if isinstance(value, self.parent.tpe):
        self.observer.onNext(value)

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

    def onCompleted(self):
      self.observer.onCompleted()
      self.dispose() class Select(Producer):
  def __init__(self, source, selector, withIndex):
    self.source = source
    self.selector = selector
    self.withIndex = withIndex

  def omega(self, selector):
    if self.selector != None and not self.withIndex:
      return Select(self.source, lambda x: selector(self.selector(x)))
    else:
      return Select(self, selector)

  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(Select.Sink, self).__init__(observer, cancel)
      self.parent = parent
      self.index = -1

    def onNext(self, value):
      try:
        result = None
        if self.parent.withIndex:
          self.index += 1
          result = self.parent.selector(value, self.index)
        else:
          result = self.parent.selector(value)
      except Exception as e:
        self.observer.onError(e)
        self.dispose()
      else:
        self.observer.onNext(result)

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

    def onCompleted(self):
      self.observer.onCompleted()
      self.dispose() class SelectMany(Producer):
  def __init__(self, source, selectorOnNext, selectorOnError, selectorOnCompleted, withIndex):
    self.source = source
    self.selectorOnNext = selectorOnNext
    self.selectorOnError = selectorOnError
    self.selectorOnCompleted = selectorOnCompleted
    self.withIndex = withIndex

  def run(self, observer, cancel, setSink):
    # if self.withIterableResult:
    #   sink = self.IterableSink(self, observer, cancel)
    #   setSink(sink)
    #   return sink.run()
    # else:
    sink = self.Sink(self, observer, cancel)
    setSink(sink)
    return sink.run()

  class Sink(rx.linq.sink.Sink):
    def __init__(self, parent, observer, cancel):
      super(SelectMany.Sink, self).__init__(observer, cancel)
      self.parent = parent
      self.index = -1

    def run(self):
      self.gate = RLock()
      self.isStopped = False
      self.group = CompositeDisposable()

      self.sourceSubscription = SingleAssignmentDisposable()
      self.group.add(self.sourceSubscription)
      self.sourceSubscription.disposable = self.parent.source.subscribeSafe(self)

      return self.group

    def onNext(self, value):
      inner = None

      try:
        if self.parent.withIndex:
          self.index += 1
          inner = self.parent.selectorOnNext(value, self.index)
        else:
          inner = self.parent.selectorOnNext(value)
      except Exception as e:
        with self.gate:
          self.observer.onError(e)
          self.dispose()

      if isinstance(inner, Observable):
        self.subscribeInner(inner)
        return

      # iterable
      try:
        for current in inner:
          self.observer.onNext(current)
      except Exception as e:
        self.observer.onError(e)
        self.dispose()

    def onError(self, exception):
      if self.parent.selectorOnError != None:
        try:
          inner = None

          if self.parent.withIndex:
            self.index += 1
            inner = self.parent.selectorOnError(exception, self.index)
          else:
            inner = self.parent.selectorOnError(exception)
        except Exception as e:
          with self.gate:
            self.observer.onError(e)
            self.dispose()
        else:
          self.subscribeInner(inner)
          self.final()
      else:
        with self.gate:
          self.observer.onError(exception)
          self.dispose()

    def onCompleted(self):
      if self.parent.selectorOnCompleted != None:
        try:
          inner = None

          if self.parent.withIndex:
            inner = self.parent.selectorOnCompleted(self.index)
          else:
            inner = self.parent.selectorOnCompleted()
        except Exception as e:
          with self.gate:
            self.observer.onError(e)
            self.dispose()
            return
        else:
          self.subscribeInner(inner)

      self.final()

    def final(self):
      self.isStopped = True

      if self.group.length == 1:
        #
        # Notice there can be a race between OnCompleted of the source and any
        # of the inner sequences, where both see _group.Count == 1, and one is
        # waiting for the lock. There won't be a double OnCompleted observation
        # though, because the call to Dispose silences the observer by swapping
        # in a NopObserver<T>.
        #
        with self.gate:
          self.observer.onCompleted()
          self.dispose()
      else:
        self.sourceSubscription.dispose()

    def subscribeInner(self, inner):
      innerSubscription = SingleAssignmentDisposable()
      self.group.add(innerSubscription)
      innerSubscription.disposable = inner.subscribeSafe(self.LockingObserver(self, innerSubscription))

    class LockingObserver(Observer):
      def __init__(self, parent, subscription):
        self.parent = parent
        self.subscription = subscription

      def onNext(self, value):
        with self.parent.gate:
          self.parent.observer.onNext(value)

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

      def onCompleted(self):
        self.parent.group.remove(self.subscription)

        if self.parent.isStopped and self.parent.group.length == 1:
          #
          # Notice there can be a race between OnCompleted of the source and any
          # of the inner sequences, where both see _group.Count == 1, and one is
          # waiting for the lock. There won't be a double OnCompleted observation
          # though, because the call to Dispose silences the observer by swapping
          # in a NopObserver<T>.
          #
          with self.parent.gate:
            self.parent.observer.onCompleted()
            self.parent.dispose()
    #end LockingObserver
  #end Sink

  # class IterableSink(rx.linq.sink.Sink):
  #   def __init__(self, parent, observer, cancel):
  #     super(SelectMany.IterableSink, self).__init__(observer, cancel)
  #     self.parent = parent
  #     self.index = -1 class SkipCount(Producer):
  def __init__(self, source, count):
    self.source = source
    self.count = count

  def omega(self, count):
    return SkipCount(self.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(SkipCount.Sink, self).__init__(observer, cancel)
      self.parent = parent
      self.remaining = parent.count

    def onNext(self, value):
      if self.remaining <= 0:
        self.observer.onNext(value)
      else:
        self.remaining -= 1

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

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

  def omega(self, duration):
    if duration < self.duration:
      duration = self.duration

    return SkipTime(self.source, self.duration)

  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(SkipTime.Sink, self).__init__(observer, cancel)
      self.parent = parent
      self.open = Atomic(False)

    def run(self):
      t = self.parent.scheduler.scheduleWithRelative(self.parent.duration, self.tick)
      d = self.parent.source.subscribeSafe(self)

      return CompositeDisposable(t, d)

    def tick(self):
      self.open.value = True

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

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

    def onCompleted(self):
      self.observer.onCompleted()
      self.dispose() class SkipWhile(Producer):
  def __init__(self, source, predicate, withIndex):
    self.source = source
    self.predicate = predicate
    self.withIndex = withIndex

  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(SkipWhile.Sink, self).__init__(observer, cancel)
      self.parent = parent
      self.running = False
      self.index = -1

    def onNext(self, value):
      if not self.running:
        try:
          if self.parent.withIndex:
            self.index += 1
            self.running = not self.parent.predicate(value, self.index)
          else:
            self.running = not self.parent.predicate(value)
        except Exception as e:
          self.observer.onError(e)
          self.dispose()
          return

      if self.running:
        self.observer.onNext(value)

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

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

  def omega(self, count):
    if self.count <= count:
      return self
    else:
      return TakeCount(self.source, 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(TakeCount.Sink, self).__init__(observer, cancel)
      self.parent = parent
      self.remaining = self.parent.count

    def onNext(self, value):
      if self.remaining > 0:
        self.remaining -= 1
        self.observer.onNext(value)

        if self.remaining == 0:
          self.observer.onCompleted()
          self.dispose()

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

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

  def omega(self, duration):
    if self.duration <= duration:
      return self
    else:
      return TakeTime(self.source, duration, self.scheduler)

  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(TakeTime.Sink, self).__init__(observer, cancel)
      self.parent = parent
      self.remaining = self.parent.count

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

      t = self.parent.scheduler.scheduleWithRelative(self.parent.duration, self.tick)
      d = self.parent.source.subscribeSafe(self)

      return CompositeDisposable(t, d)

    def tick(self):
      with self.gate:
        self.observer.onCompleted()
        self.dispose()

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

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

    def onCompleted(self):
      with self.gate:
        self.observer.onCompleted()
        self.dispose() class TakeWhile(Producer):
  def __init__(self, source, predicate, withIndex):
    self.source = source
    self.predicate = predicate
    self.withIndex = withIndex

  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(TakeWhile.Sink, self).__init__(observer, cancel)
      self.parent = parent
      self.running = True
      self.index = -1

    def onNext(self, value):
      if self.running:
        try:
          if self.parent.withIndex:
            self.index += 1
            self.running = self.parent.predicate(value, self.index)
          else:
            self.running = self.parent.predicate(value)
        except Exception as e:
          self.observer.onError(e)
          self.dispose()
          return

        if self.running:
          self.observer.onNext(value)
        else:
          self.observer.onCompleted()
          self.dispose()

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

    def onCompleted(self):
      self.observer.onCompleted()
      self.dispose() class Where(Producer):
  def __init__(self, source, predicate, withIndex):
    self.source = source
    self.predicate = predicate
    self.withIndex = withIndex

  def omega(self, predicate):
    if self.withIndex:
      return Where(self, predicate)
    else:
      return Where(self.source, lambda x: self.predicate(x) and predicate(x))

  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(Where.Sink, self).__init__(observer, cancel)
      self.parent = parent
      self.index = -1

    def onNext(self, value):
      shouldRun = False

      try:
        if self.parent.withIndex:
          self.index += 1
          shouldRun = self.parent.predicate(value, self.index)
        else:
          shouldRun = self.parent.predicate(value)
      except Exception as e:
        self.observer.onError(e)
        self.dispose()
        return

      if shouldRun:
        self.observer.onNext(value)

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

    def onCompleted(self):
      self.observer.onCompleted()
      self.dispose() def identity(x): return x 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 .defaultIfEmpty import DefaultIfEmpty
from .distinct import Distinct
from .groupBy import GroupBy
from .groupByUntil import GroupByUntil
from .groupJoin import GroupJoin
from .join import Join
from .ofType import OfType
from .select import Select
from .selectMany import SelectMany
from .skip import SkipCount, SkipTime
from .skipWhile import SkipWhile
from .take import TakeCount, TakeTime
from .takeWhile import TakeWhile
from .where import Where
from rx.internal import identity
from rx.observable import Observable
from rx.scheduler import Scheduler

  return Select(self, selector, True)
Observable.selectEnumerate = selectEnumerate

def selectMany(self, onNext, onError=None, onCompleted=None):
  assert isinstance(self, Observable)

  on = onNext
  oe = onError
  oc = onCompleted

  if not callable(onNext):
    assert isinstance(onNext, Observable)
    on = lambda _: onNext

  if onError != None and not callable(onError):
    assert isinstance(onError, Observable)
    oe = lambda _: onError

  if onCompleted != None and not callable(onCompleted):
    assert isinstance(onCompleted, Observable)
    oc = lambda: onCompleted

  return SelectMany(self, on, oe, oc, False)
Observable.selectMany = selectMany

#inspect.getfullargspec(selector) but not working for partial
def selectManyEnumerate(self, onNext, onError=None, onCompleted=None):
  assert isinstance(self, Observable)

  on = onNext
  oe = onError
  oc = onCompleted

  if not callable(onNext):
    assert isinstance(onNext, Observable)
    on = lambda _, i: onNext

  if onError != None and not callable(onError):
    assert isinstance(onError, Observable)
    oe = lambda _, i: onError

  if onCompleted != None and not callable(onCompleted):
    assert isinstance(onCompleted, Observable)
    oc = lambda _: onCompleted

  return SelectMany(self, on, oe, oc, True)
Observable.selectManyEnumerate = selectManyEnumerate

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

  if isinstance(self, SkipCount):
    return self.omega(count)

  return SkipCount(self, count)
Observable.skip = skip

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

  if isinstance(self, SkipTime):
    return self.omega(time)

  return SkipTime(self, time, scheduler)

def skipWhile(self, predicate):
  assert isinstance(self, Observable)
  assert callable(predicate)

  return SkipWhile(self, predicate, False)
Observable.skipWhile = skipWhile

def skipWhileEnumerate(self, predicate):
  assert isinstance(self, Observable)
  assert callable(predicate)

  return SkipWhile(self, predicate, True)
Observable.skipWhileEnumerate = skipWhileEnumerate

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

  if isinstance(self, TakeCount):
    return self.omega(count)

  return TakeCount(self, count)
Observable.take = take

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

  if isinstance(self, TakeTime):
    return self.omega(time)

  return TakeTime(self, time, scheduler)

def takeWhile(self, predicate):
  assert isinstance(self, Observable)
  assert callable(predicate)

  return TakeWhile(self, predicate, False)
Observable.takeWhile = takeWhile

def takeWhileEnumerate(self, predicate):
  assert isinstance(self, Observable)
  assert callable(predicate)

  return TakeWhile(self, predicate, True)
Observable.takeWhileEnumerate = takeWhileEnumerate

def where(self, predicate):
  assert isinstance(self, Observable)
  assert callable(predicate)

  if isinstance(self, Where):
    return self.omega(predicate)
  else:

    return Where(self, predicate, False)