# Path: rx/concurrency/currentthreadscheduler.py
# class Trampoline(object):
# class CurrentThreadScheduler(Scheduler):    
#     def __init__(self):
#     def enqueue(self, item):
#     def dispose(self):
#     def run(self):
#     def __init__(self):
#     def schedule(self, action, state=None):
#     def schedule_relative(self, duetime, action, state=None):
#     def schedule_absolute(self, duetime, action, state=None):
#     def schedule_required(self):
#     def ensure_trampoline(self, action):
# 
# Path: rx/autodetachobserver.py
# class AutoDetachObserver(AbstractObserver):
# 
#     def __init__(self, observer):
#         super(AutoDetachObserver, self).__init__()
#         
#         self.observer = observer
#         self.m = SingleAssignmentDisposable()
# 
#     def next(self, value):
#         try:
#             self.observer.on_next(value)
#         except Exception as ex:
#             traceback.print_exc(file=sys.stdout)
#             self.dispose()
#         
#     def error(self, exn):
#         try:
#             self.observer.on_error(exn)
#         finally:
#             self.dispose()
#         
#     def completed(self):
#         try:
#             self.observer.on_completed()
#         finally:
#             self.dispose()
#     
#     def set_disposable(self, value):
#         self.m.disposable = value
#     
#     disposable = property(fset=set_disposable)
# 
#     def dispose(self):
#         super(AutoDetachObserver, self).dispose()
#         self.m.dispose()
# 
# Path: rx/observable.py
# class Observable(object):
#     """Represents a push-style collection."""
# 
#     initializers = []
# 
#     def __init__(self, subscribe):
#         self._subscribe = subscribe
#         
#         # Run exension method initializers added by meta class
#         for init in self.initializers:
#             init(self, subscribe)
# 
#     def subscribe(self, on_next=None, on_error=None, on_completed=None):
#         """Subscribes an observer to the observable sequence. Returns the source
#         sequence whose subscriptions and unsubscriptions happen on the specified scheduler.
#         
#         1 - source.subscribe()
#         2 - source.subscribe(observer)
#         3 - source.subscribe(on_next)
#         4 - source.subscribe(on_next, on_error)
#         5 - source.subscribe(on_next, on_error, on_completed)
#         
#         Keyword arguments:
#         observer_or_on_next -- [Optional] The object that is to receive notifications or an action to invoke for each element in the observable sequence.
#         on_error -- [Optional] Action to invoke upon exceptional termination of the observable sequence.
#         on_completed -- [Optional] Action to invoke upon graceful termination of the observable sequence.
#         """
#         if isinstance(on_next, AbstractObserver):
#             observer = on_next
#         else:
#             observer = Observer(on_next, on_error, on_completed)
#             
#         return self._subscribe(observer)


from rx.concurrency import current_thread_scheduler
from .autodetachobserver import AutoDetachObserver
from .observable import Observable


class AnonymousObservable(Observable):
    """Class to create an Observable instance from a delegate-based implementation
    of the Subscribe method."""
    
    def __init__(self, subscribe):
        """Creates an observable sequence object from the specified subscription function.

        Keyword arguments:
        subscribe -- Subscribe method implementation.
        """
        def _subscribe(observer):
            auto_detach_observer = AutoDetachObserver(observer)

            # Backup. Kick queue if there's anything left: TODO: can we move this somewhere else?
            if current_thread_scheduler.schedule_required():
                def action(scheduler, state=None):
                    try:
                        auto_detach_observer.disposable = subscribe(auto_detach_observer)
                    except Exception as ex:
                        if not auto_detach_observer.fail(ex):
                            raise ex

# Path: rx/concurrency/currentthreadscheduler.py
# class Trampoline(object):
# class CurrentThreadScheduler(Scheduler):    
#     def __init__(self):
#     def enqueue(self, item):
#     def dispose(self):
#     def run(self):
#     def __init__(self):
#     def schedule(self, action, state=None):
#     def schedule_relative(self, duetime, action, state=None):
#     def schedule_absolute(self, duetime, action, state=None):
#     def schedule_required(self):
#     def ensure_trampoline(self, action):
# 
# Path: rx/autodetachobserver.py
# class AutoDetachObserver(AbstractObserver):
# 
#     def __init__(self, observer):
#         super(AutoDetachObserver, self).__init__()
#         
#         self.observer = observer
#         self.m = SingleAssignmentDisposable()
# 
#     def next(self, value):
#         try:
#             self.observer.on_next(value)
#         except Exception as ex:
#             traceback.print_exc(file=sys.stdout)
#             self.dispose()
#         
#     def error(self, exn):
#         try:
#             self.observer.on_error(exn)
#         finally:
#             self.dispose()
#         
#     def completed(self):
#         try:
#             self.observer.on_completed()
#         finally:
#             self.dispose()
#     
#     def set_disposable(self, value):
#         self.m.disposable = value
#     
#     disposable = property(fset=set_disposable)
# 
#     def dispose(self):
#         super(AutoDetachObserver, self).dispose()
#         self.m.dispose()
# 
# Path: rx/observable.py
# class Observable(object):
#     """Represents a push-style collection."""
# 
#     initializers = []
# 
#     def __init__(self, subscribe):
#         self._subscribe = subscribe
#         
#         # Run exension method initializers added by meta class
#         for init in self.initializers:
#             init(self, subscribe)
# 
#     def subscribe(self, on_next=None, on_error=None, on_completed=None):
#         """Subscribes an observer to the observable sequence. Returns the source
#         sequence whose subscriptions and unsubscriptions happen on the specified scheduler.
#         
#         1 - source.subscribe()
#         2 - source.subscribe(observer)
#         3 - source.subscribe(on_next)
#         4 - source.subscribe(on_next, on_error)
#         5 - source.subscribe(on_next, on_error, on_completed)
#         
#         Keyword arguments:
#         observer_or_on_next -- [Optional] The object that is to receive notifications or an action to invoke for each element in the observable sequence.
#         on_error -- [Optional] Action to invoke upon exceptional termination of the observable sequence.
#         on_completed -- [Optional] Action to invoke upon graceful termination of the observable sequence.
#         """
#         if isinstance(on_next, AbstractObserver):
#             observer = on_next
#         else:
#             observer = Observer(on_next, on_error, on_completed)
#             
#         return self._subscribe(observer)


# Path: rx/anonymousobservable.py
from rx.concurrency import current_thread_scheduler
from .autodetachobserver import AutoDetachObserver
from .observable import Observable



class AnonymousObservable(Observable):
    """Class to create an Observable instance from a delegate-based implementation
    of the Subscribe method."""
    
    def __init__(self, subscribe):
        """Creates an observable sequence object from the specified subscription function.

        Keyword arguments:
        subscribe -- Subscribe method implementation.
        """
        def _subscribe(observer):
            auto_detach_observer = AutoDetachObserver(observer)

            # Backup. Kick queue if there's anything left: TODO: can we move this somewhere else?
            if current_thread_scheduler.schedule_required():
                def action(scheduler, state=None):
                    try:
                        auto_detach_observer.disposable = subscribe(auto_detach_observer)
                    except Exception as ex:
                        if not auto_detach_observer.fail(ex):
                            raise ex

                current_thread_scheduler.schedule(action)