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

這可能會導致你看到線程問題。