from rx import AnonymousObservable
from rx.core import ObservableBase, Disposable
from rx.disposables import CompositeDisposable


class ConnectableObservable(ObservableBase):
    """Represents an observable that can be connected and disconnected."""

    def __init__(self, source, subject):
        self.subject = subject
        self.source = source.as_observable()
        self.has_subscription = False
        self.subscription = None

        super(ConnectableObservable, self).__init__()

    def _subscribe_core(self, observer):
        return self.subject.subscribe(observer)

    def connect(self):
        """Connects the observable."""

        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 ref_count(self):
        """Returns an observable sequence that stays connected to the
        source as long as there is at least one subscription to the
        observable sequence.
        """

        connectable_subscription = [None]
        count = [0]
        source = self

        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)
        return AnonymousObservable(subscribe)

@@ -1759,16 +1759,1020 @@
 ispose)%0A
+%0A        return AnonymousObservable(subscribe)%0A%0A    def auto_connect(self, subscriber_count=1):%0A        %22%22%22Returns an observable sequence that stays connected to the%0A        source indefinitely to the observable sequence.%0A        Providing a subscriber_count will cause it to connect() after that many subscriptions occur.%0A        A subscriber_count of 0 will result in emissions firing immediately without waiting for subscribers.%0A        %22%22%22%0A%0A        connectable_subscription = %5BNone%5D%0A        count = %5B0%5D%0A        source = self%0A%0A        if subscriber_count == 0:%0A            source.connect()%0A%0A        def subscribe(observer):%0A            count%5B0%5D += 1%0A            should_connect = count%5B0%5D == subscriber_count%0A            subscription = source.subscribe(observer)%0A            if should_connect:%0A                connectable_subscription%5B0%5D = source.connect()%0A%0A            def dispose():%0A                subscription.dispose()%0A                count%5B0%5D -= 1%0A%0A            return Disposable.create(dispose)%0A%0A
         

implement auto_connect()