# 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/disposables/serialdisposable.py
# class SerialDisposable(Disposable):
#     def __init__(self):
#         self.is_disposed = False
#         self.current = None
# 
#     def get_disposable(self):
#         return self.current
# 
#     def set_disposable(self, value):
#         should_dispose = self.is_disposed
#         old = None
# 
#         if not should_dispose:
#             old = self.current
#             self.current = value
#         
#         if old:
#             old.dispose()
#         
#         if should_dispose and value:
#             value.dispose()
#     
#     disposable = property(get_disposable, set_disposable)
# 
#     def dispose(self):
#         old = None
#         if not self.is_disposed:
#             self.is_disposed = True
#             old = self.current
#             self.current = None
#         
#         if old:
#             old.dispose()


import math
from datetime import datetime
from rx.observable import Observable
from rx.testing import TestScheduler, ReactiveTest
from rx.disposables import SerialDisposable
# Helper function for raising exceptions within lambdas
def _raise(ex):
    raise RxException(ex)

def test_select_throws():
    try:
        Observable.return_value(1) \
            .select(lambda x, y: x) \
            .subscribe(lambda x: _raise("ex"))
    except RxException:
        pass

    try:
        Observable.throw_exception('ex') \
            .select(lambda x, y: x) \
            .subscribe(on_error=lambda ex: _raise(ex))
    except RxException:
        pass

    try:
        Observable.empty() \
            .select(lambda x, y: x) \
            .subscribe(lambda x: x, lambda ex: ex, lambda: _raise('ex'))
    except RxException:
        pass

    def subscribe(observer):
        _raise('ex')

    try:

# 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/disposables/serialdisposable.py
# class SerialDisposable(Disposable):
#     def __init__(self):
#         self.is_disposed = False
#         self.current = None
# 
#     def get_disposable(self):
#         return self.current
# 
#     def set_disposable(self, value):
#         should_dispose = self.is_disposed
#         old = None
# 
#         if not should_dispose:
#             old = self.current
#             self.current = value
#         
#         if old:
#             old.dispose()
#         
#         if should_dispose and value:
#             value.dispose()
#     
#     disposable = property(get_disposable, set_disposable)
# 
#     def dispose(self):
#         old = None
#         if not self.is_disposed:
#             self.is_disposed = True
#             old = self.current
#             self.current = None
#         
#         if old:
#             old.dispose()


# Path: tests/test_standardqueryoperators/test_select.py
import math
from datetime import datetime
from rx.observable import Observable
from rx.testing import TestScheduler, ReactiveTest
from rx.disposables import SerialDisposable

# Helper function for raising exceptions within lambdas
def _raise(ex):
    raise RxException(ex)

def test_select_throws():
    try:
        Observable.return_value(1) \
            .select(lambda x, y: x) \
            .subscribe(lambda x: _raise("ex"))
    except RxException:
        pass

    try:
        Observable.throw_exception('ex') \
            .select(lambda x, y: x) \
            .subscribe(on_error=lambda ex: _raise(ex))
    except RxException:
        pass

    try:
        Observable.empty() \
            .select(lambda x, y: x) \
            .subscribe(lambda x: x, lambda ex: ex, lambda: _raise('ex'))
    except RxException:
        pass

    def subscribe(observer):
        _raise('ex')

    try:

        Observable.create(subscribe) \