Switch-case and If-else blocks in Rx-Java streams

One thing that we miss in Reactive Extensions (Rx...) is the flow control operators, if we need to do different behaviors bases on the emitted items, we have to make those ugly If-Else or Switch-Case blocks inside a flatMap() or whatever operator ... a similar case to this is as follows :

List<Integer> list = Arrays.asList(1, 2, 3, 4);

Observable.fromIterable(list)
        .flatMap(integer -> {
            switch (integer) {
                case 1:
                    return Observable.just("First Odd Number : " + integer);
                case 2:
                    return Observable.just("First Even Number : " + integer);
                default:
                    return Observable.empty();
            }
        }).subscribe(System.out::println);

Of course this sample is just for clarification, it is not about even and odd numbers ... a better way to do the same code snippet in a more readable and modular way, is to use a "SwitchCase" operator :

Map<Integer, Function<Integer, String>> caseBlocks = new HashMap<>(2);
caseBlocks.put(1, (integer) -> "First Odd Number : " + integer);
caseBlocks.put(2, (integer) -> "First Even Number : " + integer);

List<Integer> list = Arrays.asList(1, 2, 3, 4);

Observable.fromIterable(list)
        .lift(new SwitchCase<>(caseBlocks))
        .subscribe(System.out::println);

Both code snippets do the same thing, but the main difference is the "readability" , the second snippet reduced the Control Coupling between the Rx-Java stream and the outside world ... it takes a bunch of functions in a map, if any key in this map is equal to the emitted item, it will execute what ever function it finds, imagine if this map is injected from out side, like as follows :

public MyObject(Map<Integer, Function<Integer, String>> caseBlocks){

    List<Integer> list = Arrays.asList(1, 2, 3, 4);
    Observable.fromIterable(list)
            .lift(new SwitchCase<>(caseBlocks))
            .subscribe(System.out::println);
}

Now we have our Rx-Java stream free of Switch-case blocks, the stream code does not depend on external values to determine how to continue the down stream, it receives a Map and this map holds these values and functions ... we decoupled the flow control from our stream

If we want to emit the ONLY first item that matches our condition, we may implement this as follows :

Observable.fromIterable(list)
        .flatMap(integer -> {
            switch (integer) {
                case 1:
                    return Observable.just("First Odd Number : " + integer);
                case 2:
                    return Observable.just("First Even Number : " + integer);
                default:
                    return Observable.empty();
            }
        })
        .firstElement()
        .subscribe(System.out::println);

Now instead of adding extra operation, which is firstElement(), we can use "SwitchCaseBreak" :

Map<Integer, Function<Integer, String>> caseBlocks = new HashMap<>(2);
caseBlocks.put(1, (integer) -> "First Odd Number : " + integer);
caseBlocks.put(2, (integer) -> "First Even Number : " + integer);

List<Integer> list = Arrays.asList(1, 2, 3, 4);

Observable.fromIterable(list)
        .lift(new SwitchCaseBreak<>(caseBlocks))
        .subscribe(System.out::println);

The SwitchCase and SwitchCaseBreak operators source code can be fount at the following link, you can add a version that takes a Predicate as a key instead of an Object of type "T", if you need comparison other than "equals()" method.