# Path: rx/notification.py
# class Notification(object):
#   """Represents a notification to an observer."""
# 
#   KIND_NEXT = 0
#   KIND_ERROR = 1
#   KIND_COMPLETED = 2
# 
#   def __eq__(self, other):
#     if not isinstance(other, Notification):
#       return False
# 
#     return (
#       self.kind == other.kind and
#       self.value == other.value and
#       self.hasValue == other.hasValue and
#       self.exception == other.exception
#     )
# 
#   def accept(self, observerOrOnNext, onError=None, onCompleted=None):
#     """Accepts an observer or the methods onNext, onError, onCompleted and invokes
#     either onNext, onError, onCompleted depending on which type of :class:`Notification`
#     it is. This is an abstract method that is implemented by
#     :class:`OnNextNotification`, :class:`OnErrorNotification`,
#     and :class:`OnCompletedNotification`."""
#     # if observerOrOnNext == None or hasattr(observerOrOnNext, '__call__'):
#     #   observer = Observer.create(observerOrOnNext, onError, onComplete)
#     raise NotImplementedError()
# 
#   @staticmethod
#   def createOnNext(value):
#     return OnNextNotification(value)
# 
#   @staticmethod
#   def createOnError(exception):
#     return OnErrorNotification(exception)
# 
#   @staticmethod
#   def createOnCompleted():
#     return OnCompletedNotification()
# 
# 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()


from rx.notification import Notification
from rx.observable import Producer
import rx.linq.sink


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

  def dematerialize(self):
    return self.source.asObservable()

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

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

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

    def onError(self, exception):
      self.observer.onNext(Notification.createOnError(exception))
      self.observer.onCompleted()
      self.dispose()

    def onCompleted(self):

# Path: rx/notification.py
# class Notification(object):
#   """Represents a notification to an observer."""
# 
#   KIND_NEXT = 0
#   KIND_ERROR = 1
#   KIND_COMPLETED = 2
# 
#   def __eq__(self, other):
#     if not isinstance(other, Notification):
#       return False
# 
#     return (
#       self.kind == other.kind and
#       self.value == other.value and
#       self.hasValue == other.hasValue and
#       self.exception == other.exception
#     )
# 
#   def accept(self, observerOrOnNext, onError=None, onCompleted=None):
#     """Accepts an observer or the methods onNext, onError, onCompleted and invokes
#     either onNext, onError, onCompleted depending on which type of :class:`Notification`
#     it is. This is an abstract method that is implemented by
#     :class:`OnNextNotification`, :class:`OnErrorNotification`,
#     and :class:`OnCompletedNotification`."""
#     # if observerOrOnNext == None or hasattr(observerOrOnNext, '__call__'):
#     #   observer = Observer.create(observerOrOnNext, onError, onComplete)
#     raise NotImplementedError()
# 
#   @staticmethod
#   def createOnNext(value):
#     return OnNextNotification(value)
# 
#   @staticmethod
#   def createOnError(exception):
#     return OnErrorNotification(exception)
# 
#   @staticmethod
#   def createOnCompleted():
#     return OnCompletedNotification()
# 
# 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/linq/materialize.py
from rx.notification import Notification
from rx.observable import Producer
import rx.linq.sink



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

  def dematerialize(self):
    return self.source.asObservable()

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

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

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

    def onError(self, exception):
      self.observer.onNext(Notification.createOnError(exception))
      self.observer.onCompleted()
      self.dispose()

    def onCompleted(self):

      self.observer.onNext(Notification.createOnCompleted())