EndlessStreamWithListsToFilterKotlinTest

package test

import io.reactivex.Observable
import io.reactivex.observers.TestObserver
import io.reactivex.rxkotlin.toObservable
import org.junit.Before
import org.junit.Test

class EndlessStreamWithListsToFilterKotlinTest {
    val list1 = listOf(1, 2, 3, 4)
    val list2 = listOf(1, 2, 3, 4, 5)
    val list3 = listOf(1, 2, 3, 4, 5, 6)
    
    fun filterFunction(): (item: Int) -> Boolean = { item -> item % 2 == 0 }

    val expectedList1 = listOf(2, 4)
    val expectedList2 = listOf(2, 4)
    val expectedList3 = listOf(2, 4, 6)

    lateinit var endlessStream: Observable<List<Int>>

    @Before
    fun setup() {
        endlessStream = Observable.create { emitter ->
            emitter.onNext(list1)
            emitter.onNext(list2)
            emitter.onNext(list3)
            //            emitter.onComplete();  -> never completed
        }
    }

    private fun assertWorkingFine(testObserver: TestObserver<List<Int>>) {
        testObserver.assertNotComplete()
        testObserver.assertValueCount(3)
        testObserver.assertValues(expectedList1, expectedList2, expectedList3)
    }

    @Test
    fun shouldReturnFilteredListFromEndlessStreamUsingFlatMap() {

        val filteredEndlessStream = endlessStream.flatMapSingle { it.toObservable().filter(filterFunction()).toList() }

        assertWorkingFine(filteredEndlessStream.test())
    }

    @Test
    fun shouldReturnFilteredListFromEndlessStreamUsingMapRx() {

        val filteredEndlessStream = endlessStream.map { filterRx(it) }

        assertWorkingFine(filteredEndlessStream.test())
    }

    fun filterRx(toFilter: List<Int>): List<Int> {
        return toFilter.toObservable().filter(filterFunction()).toList().blockingGet()
    }

    // My preference
    @Test
    fun shouldReturnFilteredListFromEndlessStreamUsingMapNonRx() {

        val filteredEndlessStream = endlessStream.map { it.filter(filterFunction()) }

        assertWorkingFine(filteredEndlessStream.test())
    }

}

EndlessStreamWithListsToFilterTest

package test;

import org.junit.Before;
import org.junit.Test;

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

import io.reactivex.Observable;
import io.reactivex.observers.TestObserver;


public class EndlessStreamWithListsToFilterTest {

    final List<Integer> list1 = Arrays.asList(1, 2, 3, 4);
    final List<Integer> list2 = Arrays.asList(1, 2, 3, 4, 5);
    final List<Integer> list3 = Arrays.asList(1, 2, 3, 4, 5, 6);


    boolean filterFunction(int item) { return item % 2 == 0; }

    final List<Integer> expectedList1 = Arrays.asList(2, 4);
    final List<Integer> expectedList2 = Arrays.asList(2, 4);
    final List<Integer> expectedList3 = Arrays.asList(2, 4, 6);

    Observable<List<Integer>> endlessStream;

    @Before public void setup() {
        endlessStream = Observable.create(emitter -> {
            emitter.onNext(list1);
            emitter.onNext(list2);
            emitter.onNext(list3);
//            emitter.onComplete();  -> never completed
        });
    }

    private void assertWorkingFine(TestObserver<List<Integer>> testObserver) {
        testObserver.assertNotComplete();
        testObserver.assertValueCount(3);
        testObserver.assertValues(expectedList1, expectedList2, expectedList3);
    }

    // My preference
    @Test public void shouldReturnFilteredListFromEndlessStreamUsingFlatMap() {

        Observable<List<Integer>> filteredEndlessStream = endlessStream.flatMapSingle(list ->
                Observable.fromIterable(list).filter(this::filterFunction).toList()
        );

        assertWorkingFine(filteredEndlessStream.test());
    }

    @Test public void shouldReturnFilteredListFromEndlessStreamUsingMapRx() {

        Observable<List<Integer>> filteredEndlessStream = endlessStream.map(this::filterRx);

        assertWorkingFine(filteredEndlessStream.test());
    }

    List<Integer> filterRx(List<Integer> toFilter) {
        return Observable.fromIterable(toFilter).filter(this::filterFunction).toList().blockingGet();
    }

    @Test public void shouldReturnFilteredListFromEndlessStreamUsingMapNonRx() {

        Observable<List<Integer>> filteredEndlessStream = endlessStream.map(this::filterStd);

        assertWorkingFine(filteredEndlessStream.test());
    }

    List<Integer> filterStd(List<Integer> toFilter) {
        // in kotlin just: toFilter.filter {it % 2 == 0}
        List<Integer> filtered = new ArrayList<>();
        for (Integer integer : toFilter) {
            if (filterFunction(integer)) filtered.add(integer);
        }
        return filtered;
    }

}