# Path: rx/disposable.py
# class SingleAssignmentDisposable(Cancelable):
#   """Represents a disposable resource which only allows
#   a single assignment of its underlying disposable resource.
#   If an underlying disposable resource has already been set,
#   future attempts to set the underlying disposable resource
#   will throw an Error."""
# 
#   def __init__(self):
#     super(SingleAssignmentDisposable, self).__init__()
#     self.current = None
# 
#   def disposable():
#     def fget(self):
#       with self.lock:
#         if self.isDisposed:
#           return Disposable.empty()
#         else:
#           return self.current
#     def fset(self, value):
#       old = self.current
#       self.current = value
# 
#       if old != None: raise Exception("Disposable has already been assigned")
# 
#       if self.isDisposed and value != None: value.dispose()
#     return locals()
# 
#   disposable = property(**disposable())
# 
#   def dispose(self):
#     if not self._isDisposed.exchange(True):
#       if self.current != None:
#         self.current.dispose()
# 
# Path: rx/exceptions.py
# class DisposedException(Exception):
#   def __init__(self):
#     super(DisposedException, self).__init__("Object already disposed")
# 
# Path: rx/observer.py
# class Observer(Disposable):
#   """Represents the IObserver Interface.
#   Has some static helper methods attached"""
# 
#   @staticmethod
#   def create(onNext=None, onError=None, onCompleted=None):
#     if onNext == None:
#       onNext = noop
#     if onError == None:
#       onError = defaultError
#     if onCompleted == None:
#       onCompleted = noop
# 
#     return AnonymousObserver(onNext, onError, onCompleted)
# 
#   @staticmethod
#   def synchronize(observer, lock=None):
#     if lock == None:
#       lock = RLock()
# 
#     return SynchronizedObserver(observer, lock)
# 
#   @staticmethod
#   def fromNotifier(handler):
#     return AnonymousObserver(
#       lambda x: handler(Notification.createOnNext(x)),
#       lambda ex: handler(Notification.createOnError(ex)),
#       lambda: handler(Notification.createOnCompleted())
#     )
# 
#   def toNotifier(self):
#     return lambda n: n.accept(self)
# 
#   def asObserver(self):
#     return AnonymousObserver(self.onNext, self.onError, self.onCompleted)
# 
#   def checked(self):
#     return CheckedObserver(self)
# 
#   def notifyOn(self, scheduler):
#     return ScheduledObserver(scheduler, self)
# 
#   def onNext(self, value):
#     raise NotImplementedError()
# 
#   def onError(self, exception):
#     raise NotImplementedError()
# 
#   def onCompleted(self):
#     raise NotImplementedError()


from rx.disposable import SingleAssignmentDisposable
from rx.exceptions import DisposedException
from rx.observer import Observer
from queue import Empty, Queue
from threading import Semaphore
import rx.linq.sink
      raise DisposedException()

    try:
      return self.queue.get(True, 0)
    except Empty:
      pass

    if self.error != None:
      raise self.error

    assert self.done

    self.gate.release()

    raise StopIteration()

  def dispose(self):
    self.subscription.dispose()
    self.disposed = True
    self.gate.release()

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

    def onNext(self, value):
      try:

# Path: rx/disposable.py
# class SingleAssignmentDisposable(Cancelable):
#   """Represents a disposable resource which only allows
#   a single assignment of its underlying disposable resource.
#   If an underlying disposable resource has already been set,
#   future attempts to set the underlying disposable resource
#   will throw an Error."""
# 
#   def __init__(self):
#     super(SingleAssignmentDisposable, self).__init__()
#     self.current = None
# 
#   def disposable():
#     def fget(self):
#       with self.lock:
#         if self.isDisposed:
#           return Disposable.empty()
#         else:
#           return self.current
#     def fset(self, value):
#       old = self.current
#       self.current = value
# 
#       if old != None: raise Exception("Disposable has already been assigned")
# 
#       if self.isDisposed and value != None: value.dispose()
#     return locals()
# 
#   disposable = property(**disposable())
# 
#   def dispose(self):
#     if not self._isDisposed.exchange(True):
#       if self.current != None:
#         self.current.dispose()
# 
# Path: rx/exceptions.py
# class DisposedException(Exception):
#   def __init__(self):
#     super(DisposedException, self).__init__("Object already disposed")
# 
# Path: rx/observer.py
# class Observer(Disposable):
#   """Represents the IObserver Interface.
#   Has some static helper methods attached"""
# 
#   @staticmethod
#   def create(onNext=None, onError=None, onCompleted=None):
#     if onNext == None:
#       onNext = noop
#     if onError == None:
#       onError = defaultError
#     if onCompleted == None:
#       onCompleted = noop
# 
#     return AnonymousObserver(onNext, onError, onCompleted)
# 
#   @staticmethod
#   def synchronize(observer, lock=None):
#     if lock == None:
#       lock = RLock()
# 
#     return SynchronizedObserver(observer, lock)
# 
#   @staticmethod
#   def fromNotifier(handler):
#     return AnonymousObserver(
#       lambda x: handler(Notification.createOnNext(x)),
#       lambda ex: handler(Notification.createOnError(ex)),
#       lambda: handler(Notification.createOnCompleted())
#     )
# 
#   def toNotifier(self):
#     return lambda n: n.accept(self)
# 
#   def asObserver(self):
#     return AnonymousObserver(self.onNext, self.onError, self.onCompleted)
# 
#   def checked(self):
#     return CheckedObserver(self)
# 
#   def notifyOn(self, scheduler):
#     return ScheduledObserver(scheduler, self)
# 
#   def onNext(self, value):
#     raise NotImplementedError()
# 
#   def onError(self, exception):
#     raise NotImplementedError()
# 
#   def onCompleted(self):
#     raise NotImplementedError()


# Path: rx/linq/getIterator.py
from rx.disposable import SingleAssignmentDisposable
from rx.exceptions import DisposedException
from rx.observer import Observer
from queue import Empty, Queue
from threading import Semaphore
import rx.linq.sink

      raise DisposedException()

    try:
      return self.queue.get(True, 0)
    except Empty:
      pass

    if self.error != None:
      raise self.error

    assert self.done

    self.gate.release()

    raise StopIteration()

  def dispose(self):
    self.subscription.dispose()
    self.disposed = True
    self.gate.release()

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

    def onNext(self, value):
      try:

        key = self.parent.keySelector(value)