/**
 * 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
 */
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);
    }
}