title: Rx disposable not being created on time in unit tests
body: <p>I'm having trouble with testing the manual manipulation of disposables inside my class being tested. I have a <code>itemsProcessed</code> map that keeps track of the different disposables that are created, the reason is that some disposables I need to be able to dispose on demand, and others I just need to know they exist.</p>
  
  <p>My class runs fine and everything works as expected, but in my tests, I noticed that the disposables that I make aren't being inserted into my map until what would be the async code is already completed.</p>
  
  <p>I'm not sure if it has anything to do with the fact that I'm using </p>
  
  <pre><code>RxJavaPlugins.setIoSchedulerHandler { Schedulers.trampoline() }
  RxAndroidPlugins.setInitMainThreadSchedulerHandler { Schedulers.trampoline() }
  </code></pre>
  
  <p>Here are the relevant methods I'm testing, the <code>myRepo.processItem()</code> call is on a background thread</p>
  
  <pre><code>    private fun processItem(item: Item) {
          itemsProcessed[item] = myRepo.processItem(item)
                  .doOnComplete {
                      safelyDelete(item)
                      itemsProcessed.remove(item)
                  }
                  .doOnError {
                      itemsProcessed.remove(item)
                  }
                  .subscribe({}, {})
      }
  
      private fun cleanOldItems() {
          itemList.forEach {
                  if (!itemsProcessed[item].exist())
                          safelyDelete(it)
                  }
      }
  </code></pre>
  
  <p>myRepo has a processor called <code>itemProcessor</code> which calls the method above, my test is as follows </p>
  
  <pre><code>    @Test
      fun doNotDeleteItemsBeingProcessed() {
          `when`(itemProcessor.processItem(any()))
                  .thenAnswer {
                      //from my understanding of disposables, the disposable should have been made in my real class and should have been inserted into the map at this point
                      trigger cleanOldItems
                      Completable.timer(5000, TimeUnit.MILLISECONDS)
                  }
  
          repo.triggerProcessItems()
  
          Assert.assertTrue(itemList.contains(item))
      }
  </code></pre>
  
  <p>It seems like when I run the test, <code>itemsProcessed</code> map in my class is empty until the last assert line in my test is reached. When I added in <code>doOnSubscribe</code>, I noticed that <code>doOnSubscribe</code> was also called at the very end, what's causing this behaviour?</p>