# Path: rx/disposable.py
# class CompositeDisposable(Cancelable):
#   """Represents a group of disposable resources that
#   are disposed together"""
# 
#   def __init__(self, *disposables):
#     super(CompositeDisposable, self).__init__()
# 
#     if len(disposables) == 0:
#       self.disposables = []
#     elif len(disposables) == 1 and not isinstance(disposables[0], Disposable):
#       self.disposables = [d for d in disposables[0] if d != None]
#     else:
#       self.disposables = [d for d in disposables if d != None]
# 
#     self.length = len(self.disposables)
# 
#   def add(self, disposable):
#     shouldDispose = False
# 
#     with self.lock:
#       shouldDispose = self.isDisposed
# 
#       if not shouldDispose:
#         self.disposables.append(disposable)
#         self.length += 1
# 
#     if shouldDispose:
#       disposable.dispose()
# 
#   def contains(self, disposable):
#     with self.lock:
#       return self.disposables.index(disposable) >= 0
# 
#   def remove(self, disposable):
#     shouldDispose = False
# 
#     with self.lock:
#       if self.isDisposed:
#         return False
# 
#       index = self.disposables.index(disposable)
# 
#       if index < 0:
#         return False
# 
#       shouldDispose = True
# 
#       self.disposables.remove(disposable)
#       self.length -= 1
# 
#     if shouldDispose:
#       disposable.dispose()
# 
#     return shouldDispose
# 
#   def clear(self):
#     disposables = []
# 
#     with self.lock:
#       disposables = self.disposables
#       self.disposables = []
#       self.length = 0
# 
#     for disposable in disposables:
#       disposable.dispose()
# 
#   def dispose(self):
#     if not self._isDisposed.exchange(True):
#       self.clear()
# 
# class Disposable(object):
#   """Represents a disposable object"""
# 
#   def dispose(self):
#     pass
# 
#   def __enter__(self):
#     return self
# 
#   def __exit__(self, exc_type, exc_value, traceback):
#     self.dispose()
# 
#   @staticmethod
#   def create(action):
#     return AnonymouseDisposable(action)
# 
#   @staticmethod
#   def empty():
#     return disposableEmpty
# 
# Path: rx/observable.py
# 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 Producer(Observable):
#   """Base class for implementation of query operators, providing
#   performance benefits over the use of Observable.Create"""
# 
#   def subscribeCore(self, observer):
#     return self.subscribeRaw(observer, True)
# 
#   def subscribeRaw(self, observer, enableSafequard):
#     sink = SingleAssignmentDisposable()
#     subscription = SingleAssignmentDisposable()
# 
#     d = CompositeDisposable(sink, subscription)
# 
#     if enableSafequard:
#       observer = AutoDetachObserver(observer, d)
# 
#     def assignSink(s):
#       sink.disposable = s
# 
#     def scheduled():
#       subscription.disposable = self.run(observer, subscription, assignSink)
#       return Disposable.empty()
# 
#     if Scheduler.currentThread.isScheduleRequired():
#       Scheduler.currentThread.schedule(scheduled)
#     else:
#       scheduled()
# 
#     return d
# 
#   def run(self, observer, cancel, setSink):
#     raise NotImplementedError()


from rx.disposable import CompositeDisposable, Disposable
from rx.observable import Observable, Producer
import rx.linq.sink


class Using(Producer):
  def __init__(self, resourceFactory, observableFactory):
    self.resourceFactory = resourceFactory
    self.observableFactory = observableFactory

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

    def run(self):
      source = None
      disposable = Disposable.empty()

      try:
        resource = self.parent.resourceFactory()
        if resource != None:
          disposable = resource

        source = self.parent.observableFactory(resource)
      except Exception as e:
        return CompositeDisposable(Observable.throw(e).subscribeSafe(self), disposable)


# Path: rx/disposable.py
# class CompositeDisposable(Cancelable):
#   """Represents a group of disposable resources that
#   are disposed together"""
# 
#   def __init__(self, *disposables):
#     super(CompositeDisposable, self).__init__()
# 
#     if len(disposables) == 0:
#       self.disposables = []
#     elif len(disposables) == 1 and not isinstance(disposables[0], Disposable):
#       self.disposables = [d for d in disposables[0] if d != None]
#     else:
#       self.disposables = [d for d in disposables if d != None]
# 
#     self.length = len(self.disposables)
# 
#   def add(self, disposable):
#     shouldDispose = False
# 
#     with self.lock:
#       shouldDispose = self.isDisposed
# 
#       if not shouldDispose:
#         self.disposables.append(disposable)
#         self.length += 1
# 
#     if shouldDispose:
#       disposable.dispose()
# 
#   def contains(self, disposable):
#     with self.lock:
#       return self.disposables.index(disposable) >= 0
# 
#   def remove(self, disposable):
#     shouldDispose = False
# 
#     with self.lock:
#       if self.isDisposed:
#         return False
# 
#       index = self.disposables.index(disposable)
# 
#       if index < 0:
#         return False
# 
#       shouldDispose = True
# 
#       self.disposables.remove(disposable)
#       self.length -= 1
# 
#     if shouldDispose:
#       disposable.dispose()
# 
#     return shouldDispose
# 
#   def clear(self):
#     disposables = []
# 
#     with self.lock:
#       disposables = self.disposables
#       self.disposables = []
#       self.length = 0
# 
#     for disposable in disposables:
#       disposable.dispose()
# 
#   def dispose(self):
#     if not self._isDisposed.exchange(True):
#       self.clear()
# 
# class Disposable(object):
#   """Represents a disposable object"""
# 
#   def dispose(self):
#     pass
# 
#   def __enter__(self):
#     return self
# 
#   def __exit__(self, exc_type, exc_value, traceback):
#     self.dispose()
# 
#   @staticmethod
#   def create(action):
#     return AnonymouseDisposable(action)
# 
#   @staticmethod
#   def empty():
#     return disposableEmpty
# 
# Path: rx/observable.py
# 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 Producer(Observable):
#   """Base class for implementation of query operators, providing
#   performance benefits over the use of Observable.Create"""
# 
#   def subscribeCore(self, observer):
#     return self.subscribeRaw(observer, True)
# 
#   def subscribeRaw(self, observer, enableSafequard):
#     sink = SingleAssignmentDisposable()
#     subscription = SingleAssignmentDisposable()
# 
#     d = CompositeDisposable(sink, subscription)
# 
#     if enableSafequard:
#       observer = AutoDetachObserver(observer, d)
# 
#     def assignSink(s):
#       sink.disposable = s
# 
#     def scheduled():
#       subscription.disposable = self.run(observer, subscription, assignSink)
#       return Disposable.empty()
# 
#     if Scheduler.currentThread.isScheduleRequired():
#       Scheduler.currentThread.schedule(scheduled)
#     else:
#       scheduled()
# 
#     return d
# 
#   def run(self, observer, cancel, setSink):
#     raise NotImplementedError()


# Path: rx/linq/using.py
from rx.disposable import CompositeDisposable, Disposable
from rx.observable import Observable, Producer
import rx.linq.sink



class Using(Producer):
  def __init__(self, resourceFactory, observableFactory):
    self.resourceFactory = resourceFactory
    self.observableFactory = observableFactory

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

    def run(self):
      source = None
      disposable = Disposable.empty()

      try:
        resource = self.parent.resourceFactory()
        if resource != None:
          disposable = resource

        source = self.parent.observableFactory(resource)
      except Exception as e:
        return CompositeDisposable(Observable.throw(e).subscribeSafe(self), disposable)


      return CompositeDisposable(source.subscribeSafe(self), disposable)