def subscribe(observer):
        subscription = SerialDisposable()
        cancelable = SerialDisposable()

        def action(scheduler, state=None):
            try:
                source = next(sources)
            except StopIteration:
                observer.on_completed()
                return

            # Allow source to be a factory method taking an error
            source = source(state) if callable(source) else source
            current = Observable.from_future(source)

            d = SingleAssignmentDisposable()
            subscription.disposable = d

            def on_resume(state=None):
                scheduler.schedule(action, state)

            d.disposable = current.subscribe(observer.on_next, on_resume, on_resume)

        cancelable.disposable = scheduler.schedule(action)
        return CompositeDisposable(subscription, cancelable)

 def test_select_disposeinsideselector(self):
     scheduler = TestScheduler()
     xs = scheduler.create_hot_observable(on_next(100, 1), on_next(200, 2), on_next(500, 3), on_next(600, 4))
     results = scheduler.create_observer()
     d = SerialDisposable()
     invoked = [0]
     
     def projection(x, *args, **kw):
         invoked[0] += 1
         
         if scheduler.clock > 400:
             #print("*** Dispose ****")
             d.dispose()
         return x
 
     d.disposable = xs.select(projection).subscribe(results)
 
     def action(scheduler, state):
         return d.dispose()
 
     scheduler.schedule_absolute(ReactiveTest.disposed, action)
     scheduler.start()
     
     results.messages.assert_equal(on_next(100, 1), on_next(200, 2))
     xs.subscriptions.assert_equal(ReactiveTest.subscribe(0, 500))
     
     assert invoked[0] == 3

    def subscribe(observer):
        subscription = SerialDisposable()
        cancelable = SerialDisposable()
        enum = iter(sources)
        is_disposed = []

        def action(action1, state=None):
            if is_disposed:
                return

            def on_completed():
                cancelable.disposable = scheduler.schedule(action)

            try:
                current = next(enum)
            except StopIteration:
                observer.on_completed()
            except Exception as ex:
                observer.on_error(ex)
            else:
                d = SingleAssignmentDisposable()
                subscription.disposable = d
                d.disposable = current.subscribe(observer.on_next, observer.on_error, on_completed)

        cancelable.disposable = scheduler.schedule(action)

        def dispose():
            is_disposed.append(True)
        return CompositeDisposable(subscription, cancelable, Disposable.create(dispose))

def test_select_with_index_dispose_inside_selector():
    scheduler = TestScheduler()
    xs = scheduler.create_hot_observable(on_next(100, 4), on_next(200, 3), on_next(500, 2), on_next(600, 1))
    invoked = 0
    results = scheduler.create_observer()
    d = SerialDisposable()
    
    def projection(x, index):
        nonlocal invoked
        invoked += 1
        if scheduler.clock > 400:
            d.dispose()
        
        return x + index * 10

    d.disposable = xs.select(projection).subscribe(results)

    def action(scheduler, state):
        return d.dispose()

    scheduler.schedule_absolute(disposed, action)
    scheduler.start()
    results.messages.assert_equal(on_next(100, 4), on_next(200, 13))
    xs.subscriptions.assert_equal(subscribe(0, 500))
    assert invoked == 3

    def subscribe(observer):
        m = SingleAssignmentDisposable()
        d = SerialDisposable()
        d.disposable = m

        def action(scheduler, state):
            d.disposable = ScheduledDisposable(scheduler, source.subscribe(observer))

        m.disposable = scheduler.schedule(action)
        return d

class ScheduledObserver(AbstractObserver):
    def __init__(self, scheduler, observer):
        super(ScheduledObserver, self).__init__()
        self.scheduler = scheduler
        self.observer = observer
        self.is_acquired = False
        self.has_faulted = False
        self.queue = []
        self.disposable = SerialDisposable()

    def next(self, value):
        def func():
            self.observer.on_next(value)
        self.queue.append(func)
    
    def error(self, exception):
        def func():
            self.observer.on_error(exception)    
        self.queue.append(func)
                
    def completed(self):
        def func():
            self.observer.on_completed()        
        self.queue.append(func)
            
    def ensure_active(self):
        is_owner, parent = False, self
        if not self.has_faulted and len(self.queue):
            is_owner = not self.is_acquired
            self.is_acquired = True
        
        if is_owner:
            def action(action1, state):
                work = None
                if len(parent.queue):
                    work = parent.queue.pop(0)
                else:
                    parent.is_acquired = False
                    return
                
                try:
                    work()
                except Exception as ex:
                    parent.queue = []
                    parent.has_faulted = True
                    raise ex
        
                action1()
            self.disposable.disposable = self.scheduler.schedule_recursive(action)
    
    def dispose(self):
        super(ScheduledObserver, self).dispose()
        self.disposable.dispose()

        def subscribe(observer):
            delays = CompositeDisposable()
            at_end = [False]

            def done():
                if (at_end[0] and delays.length == 0):
                    observer.on_completed()

            subscription = SerialDisposable()

            def start():
                def on_next(x):
                    try:
                        delay = selector(x)
                    except Exception as error:
                        observer.on_error(error)
                        return

                    d = SingleAssignmentDisposable()
                    delays.add(d)

                    def on_next(_):
                        observer.on_next(x)
                        delays.remove(d)
                        done()

                    def on_completed():
                        observer.on_next(x)
                        delays.remove(d)
                        done()

                    d.disposable = delay.subscribe(on_next, observer.on_error,
                                                   on_completed)

                def on_completed():
                    at_end[0] = True
                    subscription.dispose()
                    done()

                subscription.disposable = source.subscribe(on_next,
                                                           observer.on_error,
                                                           on_completed)

            if not sub_delay:
                start()
            else:
                subscription.disposable(sub_delay.subscribe(
                    lambda _: start(),
                    observer.on_error,
                    start))

            return CompositeDisposable(subscription, delays)

def test_mutabledisposable_dispose():
    disp = [False]
    m = SerialDisposable()

    def action():
        disp[0] = True
    d = Disposable.create(action)
    m.disposable = d

    assert d == m.disposable
    assert not disp[0]
    m.dispose()
    assert disp[0]
    assert m.disposable == None

        def subscribe(observer):
            switched = False
            _id = 0
            
            original = SingleAssignmentDisposable()
            subscription = SerialDisposable()
            timer = SerialDisposable()
            subscription.disposable = original

            def create_timer():
                my_id = _id

                def action(scheduler, state=None):
                    nonlocal switched

                    switched = (_id == my_id)
                    timer_wins = switched
                    if timer_wins:
                        subscription.disposable = other.subscribe(observer)
                    
                timer.disposable = scheduler_method(duetime, action)

            create_timer()
            def on_next(x):
                nonlocal _id

                on_next_wins = not switched
                if on_next_wins:
                    _id += 1
                    observer.on_next(x)
                    create_timer()

            def on_error(e):
                nonlocal _id

                on_error_wins = not switched
                if on_error_wins:
                    _id += 1
                    observer.on_error(e)

            def on_completed():
                nonlocal _id
                
                on_completed_wins = not switched
                if on_completed_wins:
                    _id += 1
                    observer.on_completed()
            
            original.disposable = source.subscribe(on_next, on_error, on_completed)
            return CompositeDisposable(subscription, timer)

def test_mutabledisposable_dispose():
    disp = False
    m = SerialDisposable()
    
    def action():
        nonlocal disp
        disp = True
    d = Disposable(action)
    m.disposable = d

    assert d == m.disposable
    assert not disp
    m.dispose()
    assert disp
    assert m.disposable == None

 def __init__(self, scheduler, observer):
     super(ScheduledObserver, self).__init__()
     self.scheduler = scheduler
     self.observer = observer
     self.is_acquired = False
     self.has_faulted = False
     self.queue = []
     self.disposable = SerialDisposable()

        def subscribe(observer):
            d1 = SingleAssignmentDisposable()
            subscription = SerialDisposable()

            subscription.disposable = d1
            
            def on_error(exception):
                try:
                    result = handler(exception)
                except Exception as ex:
                    observer.on_error(ex)
                    return
                
                d = SingleAssignmentDisposable()
                subscription.disposable = d
                d.disposable = result.subscribe(observer)
            
            d1.disposable = source.subscribe(observer.on_next, on_error, observer.on_completed)
            return subscription

    def __init__(self, scheduler, observer):
        super(ScheduledObserver, self).__init__(self._next, self._error, self._completed)

        self.scheduler = scheduler
        self.observer = observer

        self.lock = Lock()
        self.is_acquired = False
        self.has_faulted = False
        self.queue = []
        self.disposable = SerialDisposable()

    def __init__(self, scheduler, period, action, state=None):
        """
        Keyword arguments:
        state -- Initial state passed to the action upon the first iteration.
        period -- Period for running the work periodically.
        action -- Action to be executed, potentially updating the state."""

        self._scheduler = scheduler
        self._state = state
        self._period = period
        self._action = action
        self._cancel = SerialDisposable()

def test_mutabledisposable_replacebeforedispose():
    disp1 = [False]
    disp2 = [False]
    m = SerialDisposable()

    def action1():
        disp1[0] = True
    d1 = Disposable.create(action1)
    m.disposable = d1

    assert d1 == m.disposable
    assert not disp1[0]

    def action2():
        disp2[0] = True
    d2 = Disposable.create(action2)
    m.disposable = d2

    assert d2 == m.disposable
    assert disp1[0]
    assert not disp2[0]

def test_mutabledisposable_replaceafterdispose():
    disp1 = [False]
    disp2 = [False]
    m = SerialDisposable()
    m.dispose()

    def action1():
        disp1[0] = True
    d1 = Disposable.create(action1)
    m.disposable = d1

    assert m.disposable == None
    assert disp1[0]

    def action2():
        disp2[0] = True
    d2 = Disposable.create(action2)
    m.disposable = d2

    assert m.disposable == None
    assert disp2[0]

def test_mutabledisposable_replaceafterdispose():
    disp1 = False
    disp2 = False
    m = SerialDisposable()
    m.dispose()

    def action1():
        nonlocal disp1
        disp1 = True
    d1 = Disposable(action1)
    m.disposable = d1

    assert m.disposable == None
    assert disp1

    def action2():
        nonlocal disp2
        disp2 = True
    d2 = Disposable(action2)
    m.disposable = d2

    m.disposable == None
    assert disp2

def test_mutabledisposable_replacebeforedispose():
    disp1 = False
    disp2 = False
    m = SerialDisposable()

    def action1():
        nonlocal disp1
        disp1 = True
    d1 = Disposable(action1)
    m.disposable = d1

    assert d1 == m.disposable
    assert not disp1

    def action2():
        nonlocal disp2
        disp2 = True
    d2 = Disposable(action2)
    m.disposable = d2

    assert d2 == m.disposable
    assert disp1
    assert not disp2

class SchedulePeriodic(object):
    """Scheduler with support for running periodic tasks. This type of
    scheduler can be used to run timers more efficiently instead of using
    recursive scheduling."""

    def __init__(self, scheduler, period, action, state=None):
        """
        Keyword arguments:
        state -- Initial state passed to the action upon the first iteration.
        period -- Period for running the work periodically.
        action -- Action to be executed, potentially updating the state."""

        self._scheduler = scheduler
        self._state = state
        self._period = period
        self._action = action
        self._cancel = SerialDisposable()

    def tick(self, scheduler, command):
        self._cancel.disposable = self._scheduler.schedule_relative(self._period, self.tick, 0)
        try:
            new_state = self._action(self._state)
        except Exception:
            self._cancel.dispose()
            raise
        else:
            if new_state is not None:  # Update state if other than None
                self._state = new_state

    def start(self):
        """Returns the disposable object used to cancel the scheduled recurring
        action (best effort).
        """

        self._cancel.disposable = self._scheduler.schedule_relative(self._period, self.tick, 0)
        return self._cancel

        def subscribe(observer):
            subscription = SerialDisposable()
            timer = SerialDisposable()
            original = SingleAssignmentDisposable()

            subscription.disposable = original

            switched = False
            _id = [0]

            def set_timer(timeout):
                my_id = _id[0]

                def timer_wins():
                    return _id[0] == my_id

                d = SingleAssignmentDisposable()
                timer.disposable = d

                def on_next(x):
                    if timer_wins():
                        subscription.disposable = other.subscribe(observer)

                    d.dispose()

                def on_error(e):
                    if timer_wins():
                        observer.on_error(e)

                def on_completed():
                    if timer_wins():
                        subscription.disposable = other.subscribe(observer)

                d.disposable = timeout.subscribe(on_next, on_error, on_completed)

            set_timer(first_timeout)

            def observer_wins():
                res = not switched
                if res:
                    _id[0] += 1

                return res

            def on_next(x):
                if observer_wins():
                    observer.on_next(x)
                    timeout = None
                    try:
                        timeout = timeout_duration_selector(x)
                    except Exception as e:
                        observer.on_error(e)
                        return

                    set_timer(timeout)

            def on_error(e):
                if observer_wins():
                    observer.on_error(e)

            def on_completed():
                if observer_wins():
                    observer.on_completed()

            original.disposable = source.subscribe(on_next, on_error, on_completed)
            return CompositeDisposable(subscription, timer)