# Path: rx/observable.py
# class PushToPullAdapter(object):
#   def __init__(self, source):
#     self.source = source
# 
#   def __iter__(self):
#     d = SingleAssignmentDisposable()
#     res = self.run(d)
#     d.disposable = self.source.subscribeSafe(res)
#     return res
# 
#   def run(self, subscription):
#     raise NotImplementedError()


from rx.observable import PushToPullAdapter
from threading import RLock
import rx.linq.sink
    self.getNewCollector = getNewCollector

  def run(self, subscription):
    sink = self.Sink(self, subscription)
    return sink

  class Sink(rx.linq.sink.PushToPullSink):
    def __init__(self, parent, subscription):
      super(Collect.Sink, self).__init__(subscription)
      self.parent = parent
      self.gate = RLock()
      self.collector = self.parent.getInitialCollector()
      self.hasFailed = False
      self.error = NotImplementedError
      self.hasCompleted = False
      self.done = False

    def onNext(self, value):
      with self.gate:
        try:
          self. collector = self.parent.merge(self.collector, value)
        except Exception as e:
          self.error = e
          self.hasFailed = True

          self.dispose()

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


# Path: rx/observable.py
# class PushToPullAdapter(object):
#   def __init__(self, source):
#     self.source = source
# 
#   def __iter__(self):
#     d = SingleAssignmentDisposable()
#     res = self.run(d)
#     d.disposable = self.source.subscribeSafe(res)
#     return res
# 
#   def run(self, subscription):
#     raise NotImplementedError()


# Path: rx/linq/collect.py
from rx.observable import PushToPullAdapter
from threading import RLock
import rx.linq.sink

    self.getNewCollector = getNewCollector

  def run(self, subscription):
    sink = self.Sink(self, subscription)
    return sink

  class Sink(rx.linq.sink.PushToPullSink):
    def __init__(self, parent, subscription):
      super(Collect.Sink, self).__init__(subscription)
      self.parent = parent
      self.gate = RLock()
      self.collector = self.parent.getInitialCollector()
      self.hasFailed = False
      self.error = NotImplementedError
      self.hasCompleted = False
      self.done = False

    def onNext(self, value):
      with self.gate:
        try:
          self. collector = self.parent.merge(self.collector, value)
        except Exception as e:
          self.error = e
          self.hasFailed = True

          self.dispose()

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


      with self.gate: