either_observable_construction.java

package com.foo;

import com.htravel.fp.Either;
import io.reactivex.Observable;

import static com.htravel.fp.Either.left;
import static com.htravel.fp.Either.right;

public class Bar {

    private static Either<String, Observable<Either<String, Integer>>> constructObservable(int maxItems) {
        return maxItems < 1 ?
                left("num items should be grater than zero!") :
                right(Observable.create(e -> {
                    int n = 0;
                    while (!e.isDisposed() && n <= maxItems)
                        e.onNext(n++ % 2 == 0 ? right(n) : left("odd not supported"));
                }));
    }

    private static <L, R> Observable<Either<L, R>> join(Either<L, Observable<Either<L, R>>> k) {
        return k.either(l -> Observable.create(e -> {
            if (!e.isDisposed()) e.onNext(left(l));
        }), r -> r);
    }


    public static void main(String... args) {
        join(constructObservable(5)).subscribe(System.out::println, Throwable::printStackTrace);
        join(constructObservable(-5)).subscribe(System.out::println, Throwable::printStackTrace);
    }
    
}
/*

Right { 1 }
Left { odd not supported }
Right { 3 }
Left { odd not supported }
Right { 5 }
Left { odd not supported }
Left { num items should be grater than zero! }

*/