gistfile1.txt

package net.test;

import org.junit.Test;

import java.util.concurrent.atomic.AtomicInteger;

import rx.Observable;
import rx.Subscriber;
import rx.functions.Func1;

import static junit.framework.TestCase.assertEquals;

/**
 * Created by mikolaj on 16.09.14.
 */
public class Test2 {

    @Test
    public void testName() throws Exception {
        final AtomicInteger count = new AtomicInteger(3);

        Observable.range(0, count.get())
                .map(THROW_ON_ODD)
                .map(i -> i) // #1
                .flatMap(Observable::just) // #2
                //.doOnNext(System.out::println) // #3
                .doOnEach(System.out::println) // #4
                .lift(OPTION_WRAP())
                .subscribe(
                        op -> {
                            System.out.println(op.toString());
                            count.decrementAndGet();
                        },
                        e -> System.out.println("It never will be printed" + e.getClass().getSimpleName()),
                        () -> System.out.println("end")
                );

        assertEquals(0, count.get());
    }

    private Func1<Integer, Integer> THROW_ON_ODD = i -> {
        if (i % 2 != 0) {
            throw new IllegalArgumentException(String.valueOf(i));
        }
        return i;
    };

    private <T> Observable.Operator<Option<T>, T> OPTION_WRAP() {
        return child -> new Subscriber<T>(child) {
            @Override
            public void onCompleted() {
                child.onCompleted();
            }

            @Override
            public void onError(Throwable e) {
                child.onNext(new Option<T>(e));
            }

            @Override
            public void onNext(T t) {
                child.onNext(new Option<>(t));
            }
        };
    }

    static class Option<T> {

        T res;
        Throwable error;
        boolean isError;

        Option(T res) {
            this.res = res;
        }

        Option(Throwable error) {
            this.error = error;
            isError = true;
        }

        @Override
        public String toString() {
            return isError ? "Error " + error.getClass().getSimpleName() : "Value: " + res.toString();
        }
    }
}