TestScheduler pour générique Observable avec RxJava

Question

J'ai commencé à utiliser TestScheduler. Tout fonctionne bien avec quelque chose comme ceci:

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

Si j'utilise ceci:

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

le test échoue parce que l'abonné est jamais appelé.

Tous les exemples que j'ai trouvés utilisent le TestScheduler avec un Observable.interval et transmettent le planificateur à la méthode d'usine comme dans le premier exemple. La raison pour laquelle je ne peux pas utiliser cette méthode est que dans l'application réelle, les observables ne sont pas aussi simples que celui-ci et je ne peux pas passer le planificateur. Je pensais que régler le Scheduler comme je le fais dans le second exemple serait bien, mais il semble que ce ne soit pas le cas.

Quelle est la bonne façon d'utiliser un TestScheduler pour des observables plus génériques?

Sans utiliser le TestScheduler je peux utiliser avec succès ces méthodes:

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

EDIT

pas de différence si j'utilise

@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

Si vous souhaitez remplacer les ordonnanceurs standard que vous peut utiliser (pour intervalle). Il est également possible d'écraser d'autres planificateurs standard.

RxJavaPlugins.setComputationSchedulerHandler(scheduler -> testScheduler); 

testScheduler serait votre TestScheduler. Après avoir configuré le plug-in, vous pouvez utiliser advanceTime comme dans test1

Exemples:

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