ObservableTestCase.java

public class FilterTestCase {

    @Test
    public void shouldFilter() throws Exception {

        final TestScheduler testScheduler = new TestScheduler();
        final Observable<Integer> two = Observable.just(2);
        final Filter filter = new Filter(two, testScheduler);
        final Observable<Integer> gtEqTwo = filter.filterGtEq(2);

        gtEqTwo.subscribe(new Action1<Integer>() {
            @Override
            public void call(Integer integer) {

                assertEquals(integer, Integer.valueOf(2));
            }
        });
        testScheduler.triggerActions();
    }
    
    public class Filter {

    private Observable<Integer> integerObservable;

    private final Scheduler scheduler;

    public Filter(final Observable<Integer> integerObservable, final Scheduler scheduler) {

        this.integerObservable = integerObservable;
        this.scheduler = scheduler;
    }

    public Observable<Integer> filterGtEq(final int limit) {

        return integerObservable.filter(new Func1<Integer, Boolean>() {
            @Override
            public Boolean call(final Integer value) {

                return value >= limit;
            }
        }).observeOn(this.scheduler);
    }

}
}