def test_groupdisposable_contains():
    d1 = Disposable.empty()
    d2 = Disposable.empty()

    g = CompositeDisposable(d1, d2)
    
    assert g.length == 2
    assert g.contains(d1)
    assert g.contains(d2)

def test_anonymousdisposable_dispose():
    disposed = [False]
    
    def action():
        disposed[0] = True

    d = Disposable(action)
    assert not disposed[0]
    d.dispose()
    assert disposed[0]

def test_anonymousdisposable_dispose():
    disposed = False
    
    def action():
        nonlocal disposed
        disposed = True

    d = Disposable(action)
    assert not disposed
    d.dispose()
    assert disposed

    def __init__(self, enable_queue=True):
        super(ControlledSubject, self).__init__(self._subscribe)

        self.subject = Subject()
        self.enable_queue = enable_queue
        self.queue = [] if enable_queue else None
        self.requested_count = 0
        self.requested_disposable = Disposable.empty()
        self.error = None
        self.has_failed = False
        self.has_completed = False
        self.controlled_disposable = Disposable.empty()

    def __subscribe(self, observer):
        self.check_disposed()
        if not self.is_stopped:
            self.observers.append(observer)
            return InnerSubscription(self, observer)

        if self.exception:
            observer.on_error(self.exception)
            return Disposable.empty()

        observer.on_completed()

        return Disposable.empty()

    def schedule_periodic(self, period, action, state=None):
        """Schedules a periodic piece of work by dynamically discovering the
        schedulers capabilities.

        Keyword arguments:
        period -- Period for running the work periodically.
        action -- Action to be executed.
        state -- [Optional] Initial state passed to the action upon the first
            iteration.

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

        scheduler = self
        seconds = self.to_relative(period)/1000.0
        if not seconds:
            return scheduler.schedule(action, state)

        def interval():
            new_state = action(scheduler, state)
            scheduler.schedule_periodic(period, action, new_state)

        log.debug("timeout: %s", seconds)
        timer = [eventlet.spawn_after(seconds, interval)]

        def dispose():
            timer[0].kill()

        return Disposable.create(dispose)

    def schedule_periodic(self, period, action, state=None):
        """Schedules an action to be executed periodically.

        Keyword arguments:
        period -- Period for running the work periodically.
        action -- {Function} Action to be executed.
        state -- [Optional] Initial state passed to the action upon the first
            iteration.

        Returns {Disposable} The disposable object used to cancel the scheduled
        action (best effort)."""

        scheduler = self
        seconds = self.to_relative(period)/1000.0
        if seconds == 0:
            return scheduler.schedule(action, state)

        def interval():
            new_state = action(state)
            scheduler.schedule_periodic(period, action, new_state)

        handle = [self.loop.call_later(seconds, interval)]

        def dispose():
            # nonlocal handle
            handle[0].cancel()

        return Disposable.create(dispose)

 def wrapped_action(self, state):
     try:
         return action(parent._get_recursive_wrapper(self), state)
     except Exception as ex:
         if not parent._handler(ex):
             raise Exception(ex)
         return Disposable.empty()

    def request(self, number):
        check_disposed(self)
        self.dispose_current_request()

        r = self._process_request(number)
        number = r["number_of_items"]
        if not r["return_value"]:
            self.requested_count = number

            def action():
                self.requested_count = 0
            self.requested_disposable = Disposable(action)

            return self.requested_disposable
        else:
            return Disposable.empty()

                def action2(scheduler1, state3):
                    if is_added:
                        group.remove(d)
                    else:
                        is_done[0] = True

                    recursive_action(state3)
                    return Disposable.empty()

            def schedule_work(_, state3):
                action(state3, inner_action)
                if is_added:
                    group.remove(d)
                else:
                    is_done[0] = True

                return Disposable.empty()

            def fix_subscriber(subscriber):
                """Fix subscriber to check for None or function returned to 
                decorate as Disposable"""
            
                if subscriber is None:
                    subscriber = Disposable.empty()
                elif type(subscriber) == types.FunctionType:
                    subscriber = Disposable(subscriber)

                return subscriber

    def connect(self):
        if not self.has_subscription:
            self.has_subscription = True

            def dispose():
                self.has_subscription = False

            disposable = self.source.subscribe(self.subject)
            self.subscription = CompositeDisposable(disposable, Disposable.create(dispose))

        return self.subscription

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

                    #print "action", scheduler1, state3
                    if is_added:
                        group.remove(d)
                    else:
                        is_done = True
                    
                    recursive_action(state)
                    return Disposable.empty()

 def subscribe(observer):
     disposable = Disposable.empty()
     try:
         resource = resource_factory()
         if resource:
             disposable = resource
         
         source = observable_factory(resource)
     except Exception as exception:
         d = Observable.throw_exception(exception).subscribe(observer)
         return CompositeDisposable(d, disposable)
     
     return CompositeDisposable(source.subscribe(observer), disposable)

        def subscribe(observer):
            count[0] += 1
            should_connect = count[0] == 1
            subscription = source.subscribe(observer)
            if should_connect:
                connectable_subscription[0] = source.connect()

            def dispose():
                subscription.dispose()
                count[0] -= 1
                if not count[0]:
                    connectable_subscription[0].dispose()

            return Disposable.create(dispose)

    def subscribe(self, observer):
        conn = self.source.publish()
        subscription = conn.subscribe(observer)
        connection = [Disposable.empty()]

        def on_next(b):
            if b:
                connection[0] = conn.connect()
            else:
                connection[0].dispose()
                connection[0] = Disposable.empty()

        pausable = self.subject.distinct_until_changed().subscribe(on_next)
        return CompositeDisposable(subscription, connection[0], pausable)

 def __subscribe(self, observer):
     self.check_disposed()
     if not self.is_stopped:
         self.observers.append(observer)
         observer.on_next(self.value)
         return InnerSubscription(self, observer)
     
     ex = self.exception
     if ex:
         observer.on_error(ex)
     else:
         observer.on_completed()
     
     return Disposable.empty()

    def _process_request(self, number_of_items):
        if self.enable_queue:
            #console.log('queue length', self.queue.length)

            while len(self.queue) >= number_of_items and number_of_items > 0:
                # console.log('number of items', number_of_items)
                self.subject.on_next(self.queue.shift())
                number_of_items -= 1

            if len(self.queue):
                return { "number_of_items": number_of_items, "return_value": True }
            else:
                return { "number_of_items": number_of_items, "return_value": False }

        if self.has_failed:
            self.subject.on_error(self.error)
            self.controlled_disposable.dispose()
            self.controlled_disposable = Disposable.empty()
        elif self.has_completed:
            self.subject.on_completed()
            self.controlled_disposable.dispose()
            self.controlled_disposable = Disposable.empty()

        return { "number_of_items": number_of_items, "return_value": False }

    def _subscribe(self, observer):
        with self.lock:
            self.check_disposed()
            if not self.is_stopped:
                self.observers.append(observer)
                return InnerSubscription(self, observer)

            ex = self.exception
            hv = self.has_value
            v = self.value

        if ex:
            observer.on_error(ex)
        elif hv:
            observer.on_next(v)
            observer.on_completed()
        else:
            observer.on_completed()

        return Disposable.empty()