Rx disposable在单元测试中没有及时创建

Question

我在测试被测试的类中的一次性物品的手动操作时遇到了问题。我有一个跟踪创建的不同一次性物品的itemsProcessed地图，原因是一些一次性物品我需要能够按需处置，而另一些我只需要知道它们的存在。

我的类运行得很好，一切都像预期的那样工作，但在我的测试中，我注意到我制作的一次性设备直到异步代码已经完成后才会被插入到我的map中。

我不确定是不是因为我用了

RxJavaPlugins.setIoSchedulerHandler { Schedulers.trampoline() }
RxAndroidPlugins.setInitMainThreadSchedulerHandler { Schedulers.trampoline() }

下面是我正在测试的相关方法，myRepo.processItem()调用在后台线程上

    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)
                }
    }

myRepo有一个名为itemProcessor的处理器，它调用上面的方法，我的测试如下

    @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))
    }

似乎当我运行测试时，类中的itemsProcessed映射是空的，直到到达测试中的最后一个断言行。当我在doOnSubscribe中添加时，我注意到doOnSubscribe在最后也被调用，是什么导致了这种行为？

Answer 1

trigger cleanOldItems是做什么的？假设这是一个同步调用，那么在返回任何一次性存储在地图中之前，它将调用您的清理函数。相反，您应该拥有

@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
                    Completable.timer(5000, TimeUnit.MILLISECONDS)
                }

        repo.triggerProcessItems()
        repo.triggerCleanOldItems()

        Assert.assertTrue(itemList.contains(item))
    }

还要记住，Completable.timer调用在内部使用计算调度器，所以除非您显式地为它提供调度器(就像在Completable.timer(5000, TimeUnit.MILLISECONDS, Schedulers.trampoline()中一样)或覆盖setComputationSchedulerHandler，否则您的订阅将在不同的线程上触发。