def subscribeCore(self, observer):
    old = None
    new = None

    while True:
      old = self.observer.value

      if old is DisposedObserver.instance:
        raise Exception("Object has been disposed")

      if old is DoneObserver.completed:
        observer.onCompleted()
        return Disposable.empty()

      if isinstance(old, DoneObserver):
        observer.onError(old.exception)
        return Disposable.empty()

      if old is NoopObserver.instance:
        new = observer
      else:
        if isinstance(old, ListObserver):
          new = old.add(observer)
        else:
          new = ListObserver((old, observer))

      current = self.observer.compareExchange(new, old)

      if old is current:
        break

    return self.Subscription(self, observer)

    def connect(self, observer):
      #
      # We connect the given observer to the subject first, before performing any kind
      # of initialization which will register an event handler. This is done to ensure
      # we don't have a time gap between adding the handler and connecting the user's
      # subject, e.g. when the ImmediateScheduler is used.
      #
      # [OK] Use of unsafe Subscribe: called on a known subject implementation.
      #
      connection = self.subject.subscribe(observer)

      self.count += 1
      if self.count == 1:
        try:
          self.initialize()
        except Exception as e:
          self.count -= 1
          connection.dispose()

          observer.onError(e)
          return Disposable.empty()

      def dispose():
        connection.dispose()

        with self.parent.gate:
          self.count -=1
          if self.count == 0:
            self.parent.scheduler.schedule(self.removeHandler.dispose)
            self.parent.session = None

      return Disposable.create(dispose)

    def invokeRec(self, scheduler, state):
      time = 0

      if self.hasResult:
        self.observer.onNext(self.result)

      try:
        if self.first:
          self.first = False
        else:
          state = self.parent.iterate(state)

        self.hasResult = self.parent.condition(state)

        if self.hasResult:
          self.result = self.parent.resultSelector(state)
          time = self.parent.timeSelector(state)
      except Exception as e:
        self.observer.onError(e)
        self.dispose()
        return Disposable.empty()

      if not self.hasResult:
        self.observer.onCompleted()
        self.dispose()
        return Disposable.empty()

      return self.parent.scheduler.scheduleWithRelativeAndState(
        state,
        time,
        self.invokeRec
      )

  def wrapper(observer):
    a = subscribe(observer)

    if isinstance(a, Disposable):
      return a
    elif callable(a):
      return Disposable.create(a)
    else:
      return Disposable.empty()

def test_Disposable_dispose():
    disposed = [False]

    def action():
        disposed[0] = True

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

    def run(self):
      srcs = list(self.parent.sources)

      N = len(srcs)

      self.queues = [None] * N
      self.isDone = [False] * N
      self.subscriptions = [None] * N
      self.gate = RLock()

      for i in range(0, N):
        self.queues[i] = deque()

      # Loop twice because subscribing could already yield
      # a value before all queues are initialized
      for i in range(0, N):
        d = SingleAssignmentDisposable()
        self.subscriptions[i] = d

        o = self.O(self, i)
        d.disposable = srcs[i].subscribeSafe(o)

      c = CompositeDisposable(self.subscriptions)

      def dispose():
        for q in self.queues:
          q.clear()

      c.add(Disposable.create(dispose))

      return c

  def run(self, sources):
    self.isDisposed = False
    self.subscription = SerialDisposable()
    self.gate = AsyncLock()
    self.stack = []
    self.length = []

    self.stack.append(iter(sources))

    try:
      length = len(sources)
    except TypeError:
      self.length.append(-1)
    else:
      self.length.append(length)

    def scheduled(continuation):
      self.recurse = continuation
      self.gate.wait(self.moveNext)

    cancel = Scheduler.tailRecursion.scheduleRecursive(scheduled)

    return CompositeDisposable(
      self.subscription,
      cancel,
      Disposable.create(lambda: self.gate.wait(self.dispose))
    )

    def scheduled(_, message):
      # time = message[0]
      notification = message[1]

      notification.accept(observer)

      return Disposable.empty()

  def subscribeCore(self, observer):
    index = len(self.subscriptions)

    self.observers.append(observer)
    self.subscriptions.append(Struct(
      subscribe=self.scheduler.now(),
      unsubscribe=0
    ))

    def scheduled(_, message):
      # time = message[0]
      notification = message[1]

      notification.accept(observer)

      return Disposable.empty()

    for m in self.messages:
      self.scheduler.scheduleWithRelativeAndState(m, m[0], scheduled)


    def dispose():
      self.observers.remove(observer)
      self.subscriptions[index].unsubscribe = self.scheduler.now()

    return Disposable.create(dispose)

  def subscribeCore(self, observer):
    d = self._subscribe(observer)

    if d == None:
      return Disposable.empty()
    else:
      return d

    def propagate(self, scheduler, currentId):
      with self.gate:
        if self.hasValue and self.resourceId == currentId:
          self.observer.onNext(self.value)
        self.hasValue = False

      return Disposable.empty()

  def scheduledSubscribe(self, scheduler, autoDetachObserver):
    try:
      autoDetachObserver.disposable = self.subscribeCore(autoDetachObserver)
    except Exception as e:
      if not autoDetachObserver.fail(e):
        raise e

    return Disposable.empty()

  def start(self):
    timer = Timer(self.interval, self._execute)

    self.timerDisposable.disposable = Disposable.create(timer.cancel)

    timer.start()

    return self.timerDisposable

  def scheduled():
    try:
      subject.onNext(action())
      subject.onCompleted()
    except Exception as e:
      subject.onError(e)

    return Disposable.empty()

    def scheduleDrain(self):
      def cancel():
        self.stopped = True
        self.stop.set()
        self.evt.release()

      self.stop.clear()
      self.cancelTimer.disposable = Disposable.create(cancel)
      self.scheduler.scheduleLongRunning(self.drainQueue)

    def run(self):
      def dispose():
        try:
          subscription.dispose()
        finally:
          self.parent.action()

      subscription = self.parent.source.subscribeSafe(self)
      return Disposable.create(dispose)

 def run(self):
   try:
     result = self.parent.eval()
   except Exception as e:
     self.observer.onError(e)
     self.dispose()
     return Disposable.empty()
   else:
     return result.subscribeSafe(self)

  def schedulerCallback(self, scheduler, state):
    with self.lock:
      if self.isAdded:
        self.group.remove(self.cancel)
      else:
        self.isDone = True

    self.run(state)

    return Disposable.empty()

  def _scheduleCore(self, state, action):
    d = SingleAssignmentDisposable()

    def scheduled():
      if not d.isDisposed:
        d.disposable = action(self, state)

    future = self.pool.submit(scheduled)
    cancel = Disposable.create(future.cancel)

    return CompositeDisposable(d, cancel)

    def timeout(self, scheduler, currentId):
      timerWins = False

      with self.gate:
        self.switched = self.currentId == currentId
        timerWins = self.switched

      if timerWins:
        self.subscription.disposable = self.parent.other.subscribeSafe(self.getForewarder())

      return Disposable.empty()