RxTest.java

package com.example.hokanla.rxplayground;

import org.junit.Before;
import org.junit.Test;
import org.mockito.Mock;
import org.mockito.MockitoAnnotations;

import rx.Observable;
import rx.Subscription;
import rx.subjects.PublishSubject;
import rx.subjects.Subject;

import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;

/**
 * Example local unit test, which will execute on the development machine (host).
 *
 * @see <a href="http://d.android.com/tools/testing">Testing documentation</a>
 */
public class ExampleUnitTest {

    public interface Doer {
        void doSomething();
    }

    @Mock private Doer mock;
    private static final int MAX_CONCURRENT_REQUESTS = 1;

    private Subject<Boolean, Boolean> pipe;

    @Before
    public void setup() {
        MockitoAnnotations.initMocks(this);
        pipe = PublishSubject.create();
    }

    // Fails with 1 invocation instead of 2
    @Test
    public void test_backPressureLatest() throws Exception {
        Subscription subscription = pipe.onBackpressureLatest()
                .flatMap(event -> Observable.just(true).doOnNext(it -> mock.doSomething()), MAX_CONCURRENT_REQUESTS)
                .subscribe();

        doAnswer(invocation -> {
            pipe.onNext(true);
            pipe.onNext(true);
            pipe.onNext(true);
            return null;
        }).when(mock).doSomething();

        pipe.onNext(true);

        verify(mock, times(2)).doSomething(); // current + latest

        subscription.unsubscribe();
    }

    // Throws a stack overflow
    @Test
    public void test_backPressureDrop() throws Exception {
        Subscription subscription = pipe.onBackpressureDrop()
                .flatMap(event -> Observable.just(true).doOnNext(it -> mock.doSomething()), MAX_CONCURRENT_REQUESTS)
                .subscribe();

        doAnswer(invocation -> {
            pipe.onNext(true);
            pipe.onNext(true);
            return null;
        }).when(mock).doSomething();
        pipe.onNext(true);

        verify(mock, times(1)).doSomething(); // current, drop others

        subscription.unsubscribe();
    }
}