RxJava2TestSample.java

package self.edu.observable;

import io.reactivex.Observable;
import io.reactivex.Scheduler;
import io.reactivex.observers.TestObserver;
import io.reactivex.schedulers.Schedulers;
import io.reactivex.schedulers.TestScheduler;
import org.junit.Assert;
import org.junit.Test;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Observer;
import java.util.concurrent.TimeUnit;

public class RxJava2TestSample {

    /*
     * Example 0.01
     */
    @Test
    public void testObservableWithTrivialWay() {
        // Arrange
        List<String> letters = Arrays.asList("A", "B", "C", "D", "E");
        List<String> results = new ArrayList<>();
        Observable<String> observable = Observable
                .fromIterable(letters)
                .zipWith(
                        Observable.range(1, Integer.MAX_VALUE),
                        (string, index) -> index + "-" + string
                );

        // Actions
        String value = observable.blockingFirst();

        // Assert
        Assert.assertEquals("1-A", value);
    }

    /*
     * Example 1
     */
    @Test
    public void testObservableWithTraditionalWay() {
        // Arrange
        List<String> letters = Arrays.asList("A", "B", "C", "D", "E");
        List<String> results = new ArrayList<>();
        Observable<String> observable = Observable
                .fromIterable(letters)
                .zipWith(
                        Observable.range(1, Integer.MAX_VALUE),
                        (string, index) -> index + "-" + string
                );

        // Actions
        observable.subscribe(results::add);

        // Assert
        Assert.assertTrue(!results.isEmpty());
        Assert.assertTrue(results.size() == 5);
        Assert.assertTrue(results.containsAll(Arrays.asList("1-A", "2-B", "3-C", "4-D", "5-E")));
    }

    /*
     * Example 2
     */
    @Test
    public void testObservableWithTestObserver() {
        // Arrange
        List<String> letters = Arrays.asList("A", "B", "C", "D", "E");
        TestObserver<String> observer = new TestObserver<>();

        Observable<String> observable = Observable
                .fromIterable(letters)
                .zipWith(
                        Observable.range(1, Integer.MAX_VALUE),
                        ((string, index) -> index + "-" + string)
                );

        // Action
        observable.subscribe(observer);

        // Assert
        observer.assertComplete();
        observer.assertNoErrors();
        observer.assertValueCount(5);
        observer.assertValues("1-A", "2-B", "3-C", "4-D", "5-E");
    }

    /*
     * Example 3
     */
    @Test
    public void testObservableInAnotherThreadWithTestObserver() {
        // Arrange
        List<String> letters = Arrays.asList("A", "B", "C", "D", "E");
        TestObserver<String> observer = new TestObserver<>();

        Observable<String> observable = Observable
                .fromIterable(letters)
                .zipWith(
                        Observable.range(1, Integer.MAX_VALUE),
                        ((string, index) -> index + "-" + string)
                )
                .subscribeOn(Schedulers.io());

        // Action
        observable.subscribe(observer);

        // --> Wait the observable completes
        observer.awaitTerminalEvent();

        // Assert
        observer.assertComplete();
        observer.assertNoErrors();
        observer.assertValueCount(5);
        observer.assertValues("1-A", "2-B", "3-C", "4-D", "5-E");
    }

    /*
     * Example 4
     */
    @Test
    public void testObservableWithException() {
        // Arrange
        List<String> letters = Arrays.asList("A", "B", "C", "D", "E");
        TestObserver<String> observer = new TestObserver<>();

        Observable<String> observable = Observable
                .fromIterable(letters)
                .zipWith(Observable.range(1, Integer.MAX_VALUE),
                        ((string, index) -> index + "-" + string)
                )
                .concatWith(Observable.error(new RuntimeException("error in Observable")));

        // Action
        observable.subscribe(observer);

        // Assert
        observer.assertError(RuntimeException.class);
        observer.assertNotComplete();
    }

    /*
     * Example 5
     */
    @Test
    public void testObservableWithTimeBased() {
        // Arrange
        TestScheduler scheduler = new TestScheduler();
        TestObserver<Long> observer = new TestObserver<>();
        Observable<Long> observable = Observable.just(1L).delay(10, TimeUnit.SECONDS, scheduler);

        // Action
        observable.subscribe(observer);
        scheduler.advanceTimeBy(10, TimeUnit.SECONDS);

        // Assert
        observer.assertValueCount(1);
    }
}