Java Streams: Combining two collections into a map

Question

I have two Collections, a list of warehouse ids and a collection of widgets. Widgets exist in multiple warehouses in varying quantities:

List<Long> warehouseIds;
List<Widget> widgets;

Here's an example defeinition of classes:

public class Widget {
    public Collection<Stock> getStocks();
}

public class Stock {
    public Long getWarehouseId();
    public Integer getQuantity();
}

I want to use the Streams API to create a Map, where the warehouse ID is the key, and the value is a list of Widgets with the smallest quantity at a particular warehouse. Because multiple widgets could have the same quantity, we return a list.

For example, Warehouse 111 has 5 qty of Widget A, 5 of Widget B, and 8 of Widget C.

Warehouse 222 has 0 qty of Widget A, 5 of Widget B, and 5 of Widget C The Map returned would have the following entries:

111 => ['WidgetA', 'WidgetB']

222 => ['WidgetA']

Starting the setup of the Map with keys seems pretty easy, but I don't know how to structure the downstream reduction:

warehouseIds.stream().collect(Collectors.groupingBy(
    Function::Identity,
    HashMap::new,
    ???...

I think the problem I'm having is reducing Widgets based on the stock warehouse Id, and not knowing how to return a Collector to create this list of Widgets. Here's how I would currently get the list of widgets with the smallest stock at a particular warehouse (represented by someWarehouseId):

widgets.stream().collect(Collectors.groupingBy(
    (Widget w)->
        w.getStocks()
        //for a specific warehouse
        .stream().filter(stock->stock.getWarehouseId()==someWarehouseId)
        //Get the quantity of stocks for a widget
        .collect(Collectors.summingInt(Stock::getQuantity)),
    //Use a tree map so the keys are sorted
    TreeMap::new,
    //Get the first entry
    Collectors.toList())).firstEntry().getValue();

Separating this into two tasks using forEach on the warehouse list would make this job easy, but I am wondering if I can do this in a 'one-liner'.

Answer 1

To tacke this problem, we need to use a more proper approach than using a TreeMap to select the values having the smallest quantities.

Consider the following approach:

• We make a Stream<Widget> of our initial widgets. We will need to do some processing on the stocks of each widget, but we'll also need to keep the widget around. Let's flatMap that Stream<Widget> into a Stream<Map.Entry<Stock, Widget>>: that new Stream will be composed of each Stock that we have, with its corresponding Widget.
• We filter those elements to only keep the Map.Entry<Stock, Widget> where the stock has a warehouseId contained in the warehouseIds list.
• Now, we need to group that Stream according to the warehouseId of each Stock. So we use Collectors.groupingBy(classifier, downstream) where the classifier returns that warehouseId.
• The downstream collector collects elements that are classified to the same key. In this case, for the Map.Entry<Stock, Widget> elements that were classified to the same warehouseId, we need to keep only those where the stock has the lowest quantity. There are no built-in collectors for this, let's use MoreCollectors.minAll(comparator, downstream) from the StreamEx library. If you prefer not to use the library, I've extracted its code into this answer and will use that.
• The comparator simply compares the quantity of each stock in the Map.Entry<Stock, Widget>. This makes sure that we'll keep elements with the lowest quantity for a fixed warehouseId. The downstream collector is used to reduce the collected elements. In this case, we only want to keep the widget, so we use Collectors.mapping(mapper, downstream) where the mapper returns the widget from the Map.Entry<Stock, Widget> and the downstream collectors collect into a list with Collectors.toList().

Sample code:

Map<Long, List<Widget>> map =
    widgets.stream()
           .flatMap(w -> w.getStocks().stream().map(s -> new AbstractMap.SimpleEntry<>(s, w)))
           .filter(e -> warehouseIds.contains(e.getKey().getWarehouseId()))
           .collect(Collectors.groupingBy(
              e -> e.getKey().getWarehouseId(),
              minAll(
                Comparator.comparingInt(e -> e.getKey().getQuantity()), 
                Collectors.mapping(e -> e.getValue(), Collectors.toList())
              )
           ));

with the following minAll collector:

public static <T, A, D> Collector<T, ?, D> minAll(Comparator<? super T> comparator, Collector<T, A, D> downstream) {
    return maxAll(comparator.reversed(), downstream);
}

public static <T, A, D> Collector<T, ?, D> maxAll(Comparator<? super T> comparator, Collector<? super T, A, D> downstream) {

    final class PairBox<U, V> {
        public U a;
        public V b;
        PairBox(U a, V b) {
            this.a = a;
            this.b = b;
        }
    }

    Supplier<A> downstreamSupplier = downstream.supplier();
    BiConsumer<A, ? super T> downstreamAccumulator = downstream.accumulator();
    BinaryOperator<A> downstreamCombiner = downstream.combiner();
    Supplier<PairBox<A, T>> supplier = () -> new PairBox<>(downstreamSupplier.get(), null);
    BiConsumer<PairBox<A, T>, T> accumulator = (acc, t) -> {
        if (acc.b == null) {
            downstreamAccumulator.accept(acc.a, t);
            acc.b = t;
        } else {
            int cmp = comparator.compare(t, acc.b);
            if (cmp > 0) {
                acc.a = downstreamSupplier.get();
                acc.b = t;
            }
            if (cmp >= 0)
                downstreamAccumulator.accept(acc.a, t);
        }
    };
    BinaryOperator<PairBox<A, T>> combiner = (acc1, acc2) -> {
        if (acc2.b == null) {
            return acc1;
        }
        if (acc1.b == null) {
            return acc2;
        }
        int cmp = comparator.compare(acc1.b, acc2.b);
        if (cmp > 0) {
            return acc1;
        }
        if (cmp < 0) {
            return acc2;
        }
        acc1.a = downstreamCombiner.apply(acc1.a, acc2.a);
        return acc1;
    };
    Function<PairBox<A, T>, D> finisher = acc -> downstream.finisher().apply(acc.a);
    return Collector.of(supplier, accumulator, combiner, finisher);
}