/**
 * an {@link ObservableOperator} that simulates an if-else in an RxJava Stream, it takes a {@link Map}
 * of {@link Predicate} as key and a {@link Function} as value ... the if any emitted item passes
 * the {@link Predicate}, this emitted item will be passed to the {@link Function} mapped to it,
 * and this item will be invoked and it's result will continue down the stream
 * <p>
 * sample code :
 * <p>
 * {@code List<Integer> list = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8);}<br>
 * {@code Map<Predicate<Integer>, Function<Integer, String>> blocks = new LinkedHashMap<>(2)}<br>
 * {@code blocks.put(i -> i % 2 == 0, i -> "Even number : " + i);}<br>
 * {@code blocks.put(i -> i % 2 != 0, i -> "Odd number : " + i);}<br>
 * <p>
 * {@code Observable.fromIterable(list)}<br>
 * {@code .lift(new IfElse<>(blocks))}<br>
 * {@code .subscribe(System.out::println);}<br>
 * <p>
 * // result :<br>
 * Odd number : 1<br>
 * Even number : 2<br>
 * Odd number : 3<br>
 * Even number : 4<br>
 * Odd number : 5<br>
 * Even number : 6<br>
 * Odd number : 7<br>
 * Even number : 8
 */
public class IfElse<T, R> implements ObservableOperator<R, T> {

    private final Map<Predicate<T>, Function<T, R>> blocks = new LinkedHashMap<>();

    /**
     * create a {@link IfElse} operator, if any {@link Function} returned {@code null}, the
     * whole operation will crash
     *
     * @param blocks the map that holds {@link Function} that are the if-else blocks
     */
    public IfElse(@NonNull Map<Predicate<T>, Function<T, R>> blocks) {
        if (blocks != null) {
            this.blocks.putAll(blocks);
        }
    }

    @Override
    public Observer<? super T> apply(final Observer<? super R> observer) throws Exception {
        return createResultObserver(observer);
    }

    private Observer<T> createResultObserver(final Observer<? super R> observer) {
        return new Observer<T>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {

            }

            @Override
            public void onNext(@NonNull T emittedItem) {

                List<Function<T, R>> validBlocks = Observable.fromIterable(blocks.keySet())
                        .filter(key -> key.test(emittedItem))
                        .map(blocks::get)
                        .toList()
                        .flatMapMaybe(Maybe::just)
                        .blockingGet();


                if (validBlocks == null) {
                    return;
                }

                try {
                    for (Function<T, R> block : validBlocks) {
                        invokeOnNext(observer, block.apply(emittedItem));
                    }
                } catch (Throwable e) {
                    onError(e);
                }
            }

            @Override
            public void onError(@NonNull Throwable e) {
                observer.onError(e);
            }

            @Override
            public void onComplete() {
                observer.onComplete();
            }
        };
    }

    private void invokeOnNext(Observer<? super R> observer, R onNextValue) {
        observer.onNext(onNextValue);
    }
}