SimpleRxJava2Test.java

package com.emirates;

import static org.hamcrest.CoreMatchers.hasItem;
import static org.hamcrest.CoreMatchers.hasItems;
import static org.hamcrest.CoreMatchers.notNullValue;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertThat;

import io.reactivex.Observable;
import io.reactivex.observers.TestObserver;
import io.reactivex.schedulers.TestScheduler;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.TimeUnit;
import org.junit.Test;

/**
 * Simple test for RxJava.
 * Source: https://www.baeldung.com/rxjava-testing
 */
public class SimpleTest {

    private static final List<String> WORDS = Arrays.asList(
            "the",
            "quick",
            "brown",
            "fox",
            "jumped",
            "over",
            "the",
            "lazy",
            "dog"
    );

    @Test
    public void testInSameThread() {
        // given:
        List<String> results = new ArrayList<>();
        Observable<String> observable = Observable.fromIterable(WORDS)
                .zipWith(Observable.range(1, Integer.MAX_VALUE),
                        (string, index) -> String.format("%2d. %s", index, string));
        // when:
        observable.subscribe(results::add);
        // then:
        assertThat(results, notNullValue());
        assertEquals(9, results.size());
        assertThat(results, hasItem(" 4. fox"));
    }

    @Test
    public void testUsingTestObserver() {
        // given:
        TestObserver<String> observer = new TestObserver<>();
        Observable<String> observable = Observable.fromIterable(WORDS)
                .zipWith(Observable.range(1, Integer.MAX_VALUE),
                        (string, index) -> String.format("%2d. %s", index, string));

        // when:
        observable.subscribe(observer);

        // then:
        observer.assertComplete();
        observer.assertNoErrors();
        observer.assertValueCount(9);
        assertThat(observer.values(), hasItem(" 4. fox"));
    }

    @Test
    public void testUsingTestObserver2() {
        List<String> letters = Arrays.asList("A", "B", "C", "D", "E");
        TestObserver<String> subscriber = new TestObserver<>();

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

        observable.subscribe(subscriber);

        subscriber.assertComplete();
        subscriber.assertNoErrors();
        subscriber.assertValueCount(5);
        assertThat(
                subscriber.values(),
                hasItems("1-A", "2-B", "3-C", "4-D", "5-E"));
    }

    @Test
    public void testException() {
        List<String> letters = Arrays.asList("A", "B", "C", "D", "E");
        TestObserver<String> subscriber = new TestObserver<>();

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

        observable.subscribe(subscriber);

        subscriber.assertError(RuntimeException.class);
        subscriber.assertNotComplete();
    }

    @Test
    public void testTimeBasedObservables() {
        List<String> letters = Arrays.asList("A", "B", "C", "D", "E");
        TestScheduler scheduler = new TestScheduler();
        TestObserver<String> subscriber = new TestObserver<>();
        Observable<Long> tick = Observable.interval(1, TimeUnit.SECONDS, scheduler);

        Observable<String> observable = Observable.fromIterable(letters)
                .zipWith(tick, (string, index) -> index + "-" + string);

        observable.subscribeOn(scheduler)
                .subscribe(subscriber);

        // Beginning
        subscriber.assertNoValues();
        subscriber.assertNotComplete();

        // Simulate time
        scheduler.advanceTimeBy(1, TimeUnit.SECONDS);

        // Check one result
        subscriber.assertNoErrors();
        subscriber.assertValueCount(1);
        subscriber.assertValues("0-A");

        // Check the rest of the results
        scheduler.advanceTimeTo(5, TimeUnit.SECONDS);

        subscriber.assertComplete();
        subscriber.assertNoErrors();
        subscriber.assertValueCount(5);
        assertThat(subscriber.values(), hasItems("1-B", "2-C", "3-D", "4-E"));
    }
}