TestObservableRaceCondition.class

package com.methodsignature.listlauncher;

import org.junit.Assert;
import org.junit.Test;

import java.util.ArrayList;
import java.util.List;

import io.reactivex.Observable;
import io.reactivex.Scheduler;
import io.reactivex.schedulers.Schedulers;

public class TestObservableRaceCondition {

    private interface SchedulersProvider {
        Scheduler io();
        Scheduler main();
    }

    private class TestSchedulersProvider implements SchedulersProvider {
        private Scheduler scheduler = Schedulers.trampoline();

        @Override
        public Scheduler io() {
            return scheduler;
        }

        @Override
        public Scheduler main() {
            return scheduler;
        }
    }

    private interface Repository {
        Observable<String> observeStrings();
    }

    private class TestRepository implements Repository {

        private List<String> data = new ArrayList<>();

        private TestRepository() {
            data.add("how");
            data.add("now");
            data.add("brown");
            data.add("cow");

        }

        @Override
        public Observable<String> observeStrings() {
            return Observable.fromIterable(data);
        }

        public List<String> getData() {
            return data;
        }
    }

    private class SlowTestRepository extends TestRepository {

        @Override
        public Observable<String> observeStrings() {
            return Observable.create(emitter -> {
                for (String string:getData()) {
                    try {
                        Thread.sleep(500);
                        emitter.onNext(string);
                    } catch (InterruptedException e) {
                        throw new RuntimeException("Unable to sleep thread", e);
                    }
                }

                emitter.onComplete();
            });
        }
    }

    private interface StringLoggingView {
        void log(String string);
    }

    private class TrackLogEventsStringLoggingView implements StringLoggingView {

        private List<String> logEvents = new ArrayList<>();

        @Override
        public void log(String string) {
            logEvents.add(string);
        }

        public List<String> getLogEvents() {
            return logEvents;
        }
    }

    private class StringLoggingPresenter {

        private StringLoggingView view;
        private Repository repository;
        private SchedulersProvider schedulersProvider;

        public StringLoggingPresenter(StringLoggingView view, Repository repository, SchedulersProvider schedulersProvider) {
            this.view = view;
            this.repository = repository;
            this.schedulersProvider = schedulersProvider;
        }

        public void start() {
            repository.observeStrings()
                    .observeOn(schedulersProvider.io())
                    .subscribeOn(schedulersProvider.main())
                    .subscribe(string -> view.log(string));
        }
    }

    @Test
    public void testDataProvidedToView() {
        TrackLogEventsStringLoggingView view = new TrackLogEventsStringLoggingView();
        TestRepository repository = new TestRepository();
        SchedulersProvider schedulersProvider = new TestSchedulersProvider();

        StringLoggingPresenter stringLoggingPresenter =
                new StringLoggingPresenter(view, repository, schedulersProvider);

        stringLoggingPresenter.start();
        Assert.assertEquals(repository.getData(), view.getLogEvents());
    }

    @Test
    public void testSlowDataProvidedToView() {
        TrackLogEventsStringLoggingView view = new TrackLogEventsStringLoggingView();
        TestRepository repository = new SlowTestRepository();
        SchedulersProvider schedulersProvider = new TestSchedulersProvider();

        StringLoggingPresenter stringLoggingPresenter =
                new StringLoggingPresenter(view, repository, schedulersProvider);

        stringLoggingPresenter.start();
        Assert.assertEquals(repository.getData(), view.getLogEvents());
    }
}

This is an example of testing an MVP presenter, mocking the view and provider, while using RxJava. Specifically, this class is designed to validate the use of a SchedulersProvider to provide a Schedulers.trampoline() Scheduler any time the presenter requests a Scheduler. The tests make use of stub TestRepository and SlowTestRepository across two tests. the SlowTestRepository returns data with a sleep before each data point is provided to the observable stream.