# Path: rx/disposable.py
# class SerialDisposable(Cancelable):
#   """Represents a disposable resource whose underlying
#   disposable resource can be replaced by
#   another disposable resource, causing automatic
#   disposal of the previous underlying disposable resource.
#   Also known as MultipleAssignmentDisposable."""
# 
#   def __init__(self):
#     super(SerialDisposable, self).__init__()
#     self.current = None
# 
#   def disposable():
#     doc = "The disposable property."
#     def fget(self):
#       with self.lock:
#         if self.isDisposed:
#           return Disposable.empty()
#         else:
#           return self.current
#     def fset(self, value):
#       old = None
#       shouldDispose = False
# 
#       with self.lock:
#         shouldDispose = self.isDisposed
# 
#         if not shouldDispose:
#           old = self.current
#           self.current = value
# 
#       if old != None:
#         old.dispose()
# 
#       if shouldDispose:
#         value.dispose()
#     return locals()
#   disposable = property(**disposable())
# 
#   def dispose(self):
#     if not self._isDisposed.exchange(True):
#       old = self.current
# 
#       self.current = None
# 
#       if old != None:
#         old.dispose()
# 
# 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/observable.py
# 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/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 SerialDisposable, SingleAssignmentDisposable
from rx.observable import Producer
from rx.observer import Observer
import rx.linq.sink
    return sink.run()

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

    def run(self):
      self.subscription = SerialDisposable()

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

      return self.subscription

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

    def onError(self, exception):
      if isinstance(exception, self.parent.exceptionType):
        result = None

        try:
          result = self.parent.handler(exception)
        except Exception as e:
          self.observer.onError(e)
          self.dispose()
          return


# Path: rx/disposable.py
# class SerialDisposable(Cancelable):
#   """Represents a disposable resource whose underlying
#   disposable resource can be replaced by
#   another disposable resource, causing automatic
#   disposal of the previous underlying disposable resource.
#   Also known as MultipleAssignmentDisposable."""
# 
#   def __init__(self):
#     super(SerialDisposable, self).__init__()
#     self.current = None
# 
#   def disposable():
#     doc = "The disposable property."
#     def fget(self):
#       with self.lock:
#         if self.isDisposed:
#           return Disposable.empty()
#         else:
#           return self.current
#     def fset(self, value):
#       old = None
#       shouldDispose = False
# 
#       with self.lock:
#         shouldDispose = self.isDisposed
# 
#         if not shouldDispose:
#           old = self.current
#           self.current = value
# 
#       if old != None:
#         old.dispose()
# 
#       if shouldDispose:
#         value.dispose()
#     return locals()
#   disposable = property(**disposable())
# 
#   def dispose(self):
#     if not self._isDisposed.exchange(True):
#       old = self.current
# 
#       self.current = None
# 
#       if old != None:
#         old.dispose()
# 
# 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/observable.py
# 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/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/catch.py
from rx.disposable import SerialDisposable, SingleAssignmentDisposable
from rx.observable import Producer
from rx.observer import Observer
import rx.linq.sink

    return sink.run()

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

    def run(self):
      self.subscription = SerialDisposable()

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

      return self.subscription

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

    def onError(self, exception):
      if isinstance(exception, self.parent.exceptionType):
        result = None

        try:
          result = self.parent.handler(exception)
        except Exception as e:
          self.observer.onError(e)
          self.dispose()
          return


        d = SingleAssignmentDisposable()