title: doOnDispose does not get called when subscribing on a background scheduler
body: <p>I am trying to bridge some existing rx code and api that uses futures. When I 
  manually dispose an observable I would expect onDispose() to be always called. It usually does but when I specify a custom scheduler it sometimes does not get called. My example:</p>
  
  <pre><code>class Work {
  
  private val disposables = CompositeDisposable()
  
  fun getFuture(): ListenableFuture&lt;String&gt; {
  
      val future = ResolvableFuture.create&lt;String&gt;()
  
      disposables.add(
  
              Observable.fromCallable {
                  try {
                      Thread.sleep(2000)
                  } catch (ex: InterruptedException) {
  
                  }
                  "1"
              }.firstOrError()
                  .onErrorReturn { "2" }
                  .doOnDispose {
                      println("disposing 1 on ${Thread.currentThread().name}")
                      //sometimes this dispose does not get called
                      future.set("2")
                  }
                  .subscribeOn(Schedulers.io())
                  .doOnDispose {
                      println("disposing 2 on ${Thread.currentThread().name}")
                      //only this dispose gets called every time
                      //future.set("2")
                  }
                  .subscribe(Consumer {
                      future.set("2")
                  })
      )
  
      return future
  }
  
  fun stop() {
      disposables.clear()
  }
  
  }
  
  @Test
  fun `doOnDispose does not get called`() {
      println("------------")
  
      for (i in 1..100) {
  
          val work = Work()
  
          val future = work.getFuture()
  
          println("Cancelling")
  
          work.stop()
  
          println("Getting ${Thread.currentThread().name}")
          val result = future.get(2, TimeUnit.SECONDS)
  
          assertEquals("2", result)
  
          println("------------")
      }
  }
  </code></pre>
  
  <p>What happens is only the second onDispose gets called every time. The one before .subscribeOn() is sometimes not called at all.  </p>