Kotlin：如何将使用Thread.sleep的测试转换为RxJava TestScheduler

Question

我正在编写一个工具测试，它检查是否将某些东西缓存到Rx缓冲区中，并且在某段时间（10秒）后，此Subject将缓冲值插入到我的房间数据库。

当我使用Thread.sleep（syncTimeInterval）时，测试是正确的。我想用TestScheduler编写相同的测试。

这是用了Thread.sleep版本（通过测试）：

@Test 
fun testMultipleLogs() { 
    val loadAllCloudCallBefore = appDatabase.logCloudCallDao().loadAll() 
    val loadAllLogNewSessionBefore = appDatabase.logNewSessionDao().loadAll() 

    assertEquals(0, loadAllCloudCallBefore.size) 
    assertEquals(0, loadAllLogNewSessionBefore.size) 


    Observable.interval(1, TimeUnit.SECONDS) 
      .take(20) 
      .subscribe { logManager.logNewSession() } 

    Observable.interval(1, TimeUnit.SECONDS) 
      .take(20) 
      .subscribe { logManager.logCloudCall("url", "callgoup") } 

    Observable.interval(1, TimeUnit.SECONDS) 
      .take(20) 
      .subscribe { logManager.logNewSession() } 

    Observable.interval(1, TimeUnit.SECONDS) 
      .take(20) 
      .subscribe { logManager.logCloudCall("url", "callgoup") } 

    Observable.interval(1, TimeUnit.SECONDS) 
      .take(20) 
      .subscribe { logManager.logNewSession() } 

    Observable.interval(1, TimeUnit.SECONDS) 
      .take(20) 
      .subscribe { logManager.logCloudCall("url", "callgoup") } 


    Thread.sleep(30000) 

    val loadAllCloudCallAfter = appDatabase.logCloudCallDao().loadAll() 
    val loadAllLogNewSessionAfter = appDatabase.logNewSessionDao().loadAll() 

    assertEquals(60, loadAllCloudCallAfter.size) 
    assertEquals(60, loadAllLogNewSessionAfter.size) 
} 

而在这里，这个测试没有通过，预计到TestScheduler提前时间后，大小为0（不是60）

@Test 
fun testMultipleLogs() { 
    var testScheduler: TestScheduler = TestScheduler() 

    val loadAllCloudCallBefore = appDatabase.logCloudCallDao().loadAll() 
    val loadAllLogNewSessionBefore = appDatabase.logNewSessionDao().loadAll() 

    assertEquals(0, loadAllCloudCallBefore.size) 
    assertEquals(0, loadAllLogNewSessionBefore.size) 


    Observable.interval(1, TimeUnit.SECONDS, testScheduler) 
      .take(20) 
      .subscribe { logManager.logNewSession() } 

    Observable.interval(1, TimeUnit.SECONDS, testScheduler) 
      .take(20) 
      .subscribe { logManager.logCloudCall("url", "callgoup") } 

    Observable.interval(1, TimeUnit.SECONDS, testScheduler) 
      .take(20) 
      .subscribe { logManager.logNewSession() } 

    Observable.interval(1, TimeUnit.SECONDS, testScheduler) 
      .take(20) 
      .subscribe { logManager.logCloudCall("url", "callgoup") } 

    Observable.interval(1, TimeUnit.SECONDS, testScheduler) 
      .take(20) 
      .subscribe { logManager.logNewSession() } 

    Observable.interval(1, TimeUnit.SECONDS, testScheduler) 
      .take(20) 
      .subscribe { logManager.logCloudCall("url", "callgoup") } 


    testScheduler.advanceTimeBy(21, TimeUnit.SECONDS) 

    val loadAllCloudCallAfter = appDatabase.logCloudCallDao().loadAll() 
    val loadAllLogNewSessionAfter = appDatabase.logNewSessionDao().loadAll() 

    assertEquals(60, loadAllCloudCallAfter.size) 
    assertEquals(60, loadAllLogNewSessionAfter.size) 
} 

如何正确测试这种情况？有没有办法？

UPDATE

在日志管理功能如下：

fun logCloudCall(url: String, callGroup: String) { 
    val logCloudCall = LogCloudCall(url = url, callGroup = callGroup, date = Converter.GENERAL_DATE_FORMAT.format(Date())) 

    Log.v("LogManager", logCloudCall.toString()) 
    addLog(logCloudCall) 
} 

    fun logNewSession() { 
    val logNewSession = 
      LogNewSession(
        date = Converter.GENERAL_DATE_FORMAT.format(Date())) 
    Log.v("LogManager", logNewSession.toString()) 

    addLog(logNewSession) 
} 

    fun addLog(logEvent: LogEvent) { 
    source.onNext(logEvent) 
} 

这是我在我的日志管理初始化使用机制：

val source = PublishSubject.create<LogEvent>().toSerialized() 

var logRepository: LogRepository 

init { 
    logRepository = LogRepositoryImpl(context) 
    configureSubject() 
} 


fun configureSubject() { 
    source 
      .buffer(10, TimeUnit.SECONDS) 
      .subscribe { bufferedData -> proceedValues(bufferedData) } 
} 

Answer 1

下测试通过：

@Test 
fun foo() { 
    val testScheduler = TestScheduler() 
    var count = 0 

    Observable.interval(1, TimeUnit.SECONDS, testScheduler) 
      .take(20) 
      .subscribe { count++ } 

    testScheduler.advanceTimeBy(21, SECONDS) 

    assert(count == 20) 
} 

也就是说，您的测试代码看起来正确，但结果不正确。这里唯一未知的是logManager代码。该班有线程吗？这也许可以解释为什么计数仍然是0：您可能有竞争条件。

---

这可能是由于buffer调用。 buffer内部使用的计算Scheduler：

public final Observable<List<T>> buffer(long timespan, TimeUnit unit) { 
    return buffer(timespan, unit, Schedulers.computation(), Integer.MAX_VALUE); 
} 

这可能会导致你看到线程问题。