Testing timeout handling TestSubject and TestScheduler

[Olve Hansen]

public static <T extends PlayEvent> void timeoutAndDistinctViewingSession(

            Observable<T> source, Observer<T> sink

    ) {

        source.timeout(/*Timeout after delay after last event,

         and preventing memory leaks so that inactive streams are closed */

                       3, TimeUnit.HOURS

        )

              .distinctUntilChanged(/*So that consecutive identical play events are skipped*/

                                    PlayEvent::getViewingSession

              )

              .serialize()

              .subscribe(sink /*Pass event on to function/class responsible for deciding if this is  a kick*/);

    }

I have this Observable setup, and I want to test how my "sink" behaves when the source times out. I try to do this with this unit test:

@Test

    public void testTimeoutWindow() throws Exception {

        final TestScheduler testScheduler = Schedulers.test();

        TestSubject<PlayEvent> testSubject = TestSubject.create(testScheduler);

        PlayEventRxFunctions.timeoutAndDistinctViewingSession(testSubject, this);

        final int userId = 1;

        testSubject.onNext(createPlayEvent(userId, UUID.randomUUID()));

        testScheduler.advanceTimeBy(4, TimeUnit.HOURS);

        testSubject.onNext(createPlayEvent(userId, UUID.randomUUID()));

        Assertions.assertThat(getOnErrorEvents().size()).isEqualTo(1);

        Assertions.assertThat(getOnNextEvents().size()).isEqualTo(1);

    }

Although it does not work when I advance the time using test scheduler.

 am very new to RxJava, and this might be a big misunderstanding on how things should work on my part...

Would be grateful for any pointers on how to test this setup could be tested!

Regards!

[Ben Christensen]

The TestScheduler needs to be passed into the timeout operator as well for it to use virtual time. 

Ben Christensen

310.782.5511

@benjchristensen

[Olve Hansen]

Of course, now that is really obvious. Thanks! :-)