用於RxJava的泛型Observable的TestScheduler

Question

我開始使用TestScheduler。如果我用這個代替

@Test 
public void test1() throws Exception { 
    //when 
    TestScheduler scheduler = new TestScheduler(); 
    TestObserver<Long> subscriber = new TestObserver<>(); 

    //run 
    Observable 
      .interval(1L, TimeUnit.SECONDS, scheduler) 
      .subscribeWith(subscriber); 

    //check 
    scheduler.advanceTimeBy(200, TimeUnit.SECONDS); 
    assertEquals(200, subscriber.valueCount()); 
} 

：

@Test 
public void test2() throws Exception { 
    //when 
    TestScheduler scheduler = new TestScheduler(); 
    TestObserver<Long> subscriber = new TestObserver<>(); 

    //run 
    Observable 
      .interval(1L, TimeUnit.SECONDS) 
      .observeOn(scheduler) 
      .subscribeOn(scheduler) 
      .subscribeWith(subscriber); 

    //check 
    scheduler.advanceTimeBy(200, TimeUnit.SECONDS); 
    assertEquals(200, subscriber.valueCount()); 
} 

測試失敗，因爲用戶沒有被調用一切正常，像這樣的東西。

我找到的所有例子都使用TestScheduler和Observable.interval，並將調度器傳遞給工廠方法，就像我在第一個例子中那樣。 我不能使用這種方法的原因是，在真正的應用程序中，observables並不像這樣簡單，我無法通過調度程序。 我認爲設置Scheduler就像我在第二個例子中做的那樣很好，但看起來不是。

對於更通用的Observable，使用TestScheduler的正確方法是什麼？

如果不使用TestScheduler我能成功使用這些方法：如果你想覆蓋標準的調度

@Test 
public void test3() throws Exception { 
    //when 
    Scheduler trampoline = Schedulers.trampoline(); 

    //run 
    TestObserver<Long> test = Observable 
      .interval(1L, TimeUnit.SECONDS) 
      .observeOn(trampoline) 
      .subscribeOn(trampoline) 
      .test(); 

    //check 
    test.await(3100,TimeUnit.MILLISECONDS); 
    assertEquals(3, test.valueCount()); 
} 

@Test 
public void test4() throws Exception { 
    //when 
    Scheduler trampoline = Schedulers.trampoline(); 

    //run 
    TestObserver<Long> test = Observable 
      .fromArray(1L, 2L, 3L) 
      .subscribeOn(trampoline) 
      .observeOn(trampoline) 
      .test(); 

    //check 
    assertEquals(3, test.valueCount()); 
} 

@Test 
public void test5() throws Exception { 
    //when 
    Scheduler trampoline = Schedulers.trampoline(); 

    //run 
    TestObserver<Long> test = Observable 
      .fromArray(1L, 2L, 3L) 
      .subscribeOn(trampoline) 
      .observeOn(AndroidSchedulers.mainThread()) 
      .test(); 

    //check 
    test.awaitTerminalEvent(); 
    assertEquals(3, test.valueCount()); 
} 

編輯

沒有什麼區別，如果我使用

@BeforeClass 
public static void setupClass() { 
    mScheduler = new TestScheduler(); 

    RxAndroidPlugins.setInitMainThreadSchedulerHandler(__ -> mScheduler); 
    RxJavaPlugins.setIoSchedulerHandler(__ -> mScheduler); 
} 

@Test 
public void test2() throws Exception { 
    //when 
    TestObserver<Long> subscriber = new TestObserver<>(); 

    //run 
    Observable 
      .interval(1L, TimeUnit.SECONDS) 
      .observeOn(mScheduler) 
      .subscribeOn(mScheduler) 
      .subscribeWith(subscriber); 

    //check 
    mScheduler.advanceTimeBy(200, TimeUnit.SECONDS); 
    assertEquals(200, subscriber.valueCount()); 
} 

Answer 1

你可以使用（間隔）。也可以覆蓋其他標準調度程序。

RxJavaPlugins.setComputationSchedulerHandler(scheduler -> testScheduler); 

testScheduler將是你TestScheduler。設置插件後，您可以使用advanceTime作爲test1的

例子：

@Test 
// fails because interval schedules on different thread then JUnit-Runner-thread -> fall through 
void notWorkingTest1() throws Exception { 
    TestScheduler scheduler = new TestScheduler(); 
    TestObserver<Long> subscriber = new TestObserver<>(); 

    Observable 
      .interval(1L, TimeUnit.SECONDS) 
      .observeOn(scheduler) 
      .subscribeOn(scheduler) 
      .subscribeWith(subscriber); 

    //check 
    scheduler.advanceTimeBy(200, TimeUnit.SECONDS); 
    assertEquals(200, subscriber.valueCount()); 
} 


@Test 
// not working because interval will not be scheduled on virtual time -> JUnit-Runner-Thread will close because test observable emits on different thread 
void notWorkingTest2() throws Exception { 
    //when 
    TestScheduler scheduler = new TestScheduler(); 

    //run 
    TestObserver<Long> test = Observable 
      .interval(1L, TimeUnit.SECONDS) 
      .observeOn(scheduler) 
      .subscribeOn(scheduler) 
      .test(); 

    scheduler.advanceTimeBy(200, TimeUnit.SECONDS); 

    test.assertValueCount(200); 
} 

@Test 
// runs sync. -> no JUnit-Runner-thread blocking needed 
void workingTest() throws Exception { 
    TestScheduler scheduler = new TestScheduler(); 
    RxJavaPlugins.setComputationSchedulerHandler(s -> scheduler); 

    TestObserver<Long> test = Observable 
      .interval(1L, TimeUnit.SECONDS) // executed on Schedulers.computation() 
      .observeOn(scheduler) 
      .subscribeOn(scheduler) 
      .test(); 

    scheduler.advanceTimeBy(200, TimeUnit.SECONDS); 

    test.assertValueCount(200); 
}