CartProcessor.java

package com.sandbox;

import javafx.beans.*;
import rx.Observable;
import rx.Scheduler;
import rx.functions.Action2;
import rx.functions.Func0;
import rx.schedulers.Schedulers;

import java.util.*;
import java.util.concurrent.*;
import java.util.function.BiFunction;
import java.util.stream.Collectors;
import java.util.stream.Stream;

public class CartProcessor {

    public static void main(String[] args) {
//        new CartProcessor(1L);
//        new CartProcessor(2L);
//        new CartProcessor(3L);
        /*
    <dependency>
        <groupId>io.reactivex</groupId>
        <artifactId>rxjava</artifactId>
        <version>1.3.0</version>
    </dependency>
         */
        CartProcessor processor = new CartProcessor();
    }

    public CartProcessor() {
        long cartId = 1L;
        final ExecutorService executor = Executors.newCachedThreadPool();
        final Service service = new MockService();

        Scheduler scheduler = Schedulers.from(executor);

        Observable.from(executor.submit(() -> service.getCartByUserId(cartId)))
                .subscribeOn(scheduler)
                .onErrorReturn(e -> new MissingCart())
                .subscribe(cart -> Observable.from(cart.toArray())
                    .flatMap(itemId -> Observable.just(itemId)
                            .subscribeOn(scheduler)
                            .map(id -> service.getItemById((Long) id))
                            .onErrorReturn(e -> new MissingItem()))
                    .collect(ArrayList::new, (acc, item) -> acc.add(item))
                    .subscribe(list -> System.out.println(list)));
    }

    public CartProcessor(long cartId) {

        final Service service = new MockService();

        CompletableFuture<Cart> getCartFuture = CompletableFuture
                .supplyAsync(() -> service.getCartByUserId(cartId))
                .handle((res, ex) -> {
                    if (ex != null) {
                        System.out.println(ex.getMessage());
                        return new MissingCart();
                    }
                    return res;
                });

        CompletableFuture<?> getItemsFuture = getCartFuture.thenApply(cart -> {
            if (cart instanceof MissingCart) {
                return Collections.<Item>emptyList();
            }
            Stream.Builder<CompletableFuture> combined = Stream.builder();
            cart.stream().map(itemId -> CompletableFuture
                    .supplyAsync(() -> service.getItemById(itemId))
                    .handle(missingItemHandle))
                    .forEach(combined::add);
            return combined.build()
                    .map(CompletableFuture::join)
                    .filter(item -> !(item instanceof MissingItem))
                    .collect(Collectors.toCollection(ArrayList::new));
        });

        getItemsFuture.thenAccept(System.out::println);
    }

    private BiFunction<Item, Throwable, Item> missingItemHandle = (i, t) -> {
        if (t != null) {
            return new MissingItem();
        }
        return i;
    };

    static class Item {
        Long itemId;
        String productName;

        public Item() { }

        public Item(long l, String name) {
            this.itemId = l;
            this.productName = name;
        }

        public String toString() {
            return "Item{" + "itemId=" + itemId + ", productName='" + productName + '\'' + '}';
        }
    }

    class MissingItem extends Item { }

    class Cart extends ArrayList<Long> { }

    class MissingCart extends Cart { }

    interface Service {
        Cart getCartByUserId(Long userId);

        Item getItemById(Long itemId);
    }

    class MockService implements Service {

        private Map<Long, Cart> cartMap = new HashMap<>();
        private Map<Long, Item> itemMap = new HashMap<>();

        public MockService() {
            Item item1 = new Item(1L, "soap");
            Item item2 = new Item(2L, "mop");
            Item item3 = new Item(3L, "rice");
            Item item4 = new Item(4L, "drink");
            itemMap.put(item1.itemId, item1);
            itemMap.put(item2.itemId, item2);
            itemMap.put(item3.itemId, item3);
            itemMap.put(item4.itemId, item4);
            Cart user1Cart = new Cart();
            user1Cart.add(item1.itemId);
            user1Cart.add(item2.itemId);
            user1Cart.add(item3.itemId);
            user1Cart.add(item4.itemId);
            cartMap.put(1L, user1Cart);

            Cart user2Cart = new Cart();
            user2Cart.add(item1.itemId);
            user2Cart.add(item2.itemId);
            user2Cart.add(item3.itemId);
            user2Cart.add(item4.itemId);
            user2Cart.add(5L); //non existent
            cartMap.put(2L, user2Cart);
        }

        public Cart getCartByUserId(Long userId) {
            if (!cartMap.containsKey(userId)) {
                throw new RuntimeException("Cart not found");
            }
            return cartMap.get(userId);
        }

        public Item getItemById(Long itemId) {
            if (!itemMap.containsKey(itemId)) {
                throw new RuntimeException("Item not found");
            }
            return itemMap.get(itemId);
        }
    }
}