Stream a collection and collect into multiple results depending on independent predicates

Question

I am staring at some imperative code which I am trying to convert to a purely functional style. Basically there is a iterative for-loop over inputSet in which I check 3 predicates and fill 3 outputSets depending on which predicate matches. The output sets can be overlapping. How can I do this in a pure functional way using Java 8 streams/map/filter/ etc.?

Answer 1

The easiest solution (except leaving everything as is which is even easier) is to create three separate streams:

Set<MyObj> set1 = inputSet.stream().filter(pred1).collect(Collectors.toSet());
Set<MyObj> set2 = inputSet.stream().filter(pred2).collect(Collectors.toSet());
Set<MyObj> set3 = inputSet.stream().filter(pred3).collect(Collectors.toSet());

If you have a List of predicates, you can create a corresponding List of sets as a result:

List<Predicate<MyObj>> predicates = Arrays.asList(pred1, pred2, pred3);
List<Set<MyObj>> result = predicates.stream()
        .map(pred -> inputSet.stream().filter(pred).collect(Collectors.toSet()))
        .collect(Collectors.toList());

Here the first set in the resulting list corresponds to the first predicate and so on.

If you really want to process your input in single pass (for whatever reason), you may write a special collector for this. Here's one which is quite universal:

public static <T, A, R> Collector<T, ?, List<R>> multiClassify(
        List<Predicate<T>> predicates, Collector<? super T, A, R> downstream) {
    Supplier<A> dsSupplier = downstream.supplier();
    BiConsumer<A, ? super T> dsAccumulator = downstream.accumulator();
    BinaryOperator<A> dsCombiner = downstream.combiner();

    Supplier<List<A>> supplier = () -> Stream.generate(dsSupplier)
            .limit(predicates.size()).collect(Collectors.toList());

    BiConsumer<List<A>, T> accumulator = (list, t) -> IntStream
            .range(0, predicates.size()).filter(i -> predicates.get(i).test(t))
            .forEach(i -> dsAccumulator.accept(list.get(i), t));

    BinaryOperator<List<A>> combiner = (l1, l2) -> IntStream.range(0, predicates.size())
            .mapToObj(i -> dsCombiner.apply(l1.get(i), l2.get(i)))
            .collect(Collectors.toList());

    Characteristics[] dsCharacteristics = downstream.characteristics().toArray(
            new Characteristics[0]);
    if (downstream.characteristics().contains(Characteristics.IDENTITY_FINISH)) {
        @SuppressWarnings("unchecked")
        Collector<T, ?, List<R>> result = (Collector<T, ?, List<R>>) (Collector<T, ?, ?>) 
            Collector.of(supplier, accumulator, combiner, dsCharacteristics);
        return result;
    }
    Function<A, R> dsFinisher = downstream.finisher();
    Function<List<A>, List<R>> finisher = l -> l.stream().map(dsFinisher)
           .collect(Collectors.toList());
    return Collector.of(supplier, accumulator, combiner, finisher, dsCharacteristics);
}

It takes a list of predicates and returns list of downstream collector results for each predicate. Usage example:

List<String> input = asList("abc", "ade", "bcd", "cc", "cdac");

List<Predicate<String>> preds = asList(
        s -> s.length() == 3, 
        s -> s.startsWith("a"), 
        s -> s.endsWith("c"));
List<Set<String>> result = input.stream().collect(multiClassify(preds, Collectors.toSet()));
// [[bcd, abc, ade], [abc, ade], [cc, abc, cdac]]

Answer 2

Another approach would be to use the Consumer.andThen(anotherConsumer) method to create a composed consumer made of inner consumers that execute in sequence. Each one of these inner consumers would test every predicate and classify elements depending on whether they match or not.

public static <T> Consumer<T> classify(Predicate<T> predicate, Consumer<T> action) {
    return elem -> Optional.ofNullable(elem)
        .filter(predicate)
        .ifPresent(action);
}

This utility method returns a consumer that will execute the given action on the element being consumed, as long as the predicate returns true for the element.

Test:

Stream<Integer> stream = Stream.of(1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12);

Set<Integer> set1 = new LinkedHashSet<>();
Set<Integer> set2 = new LinkedHashSet<>();
Set<Integer> set3 = new LinkedHashSet<>();

// Here's the composed consumer, made of inner consumers
Consumer<Integer> multiClassifier = classify(n -> n % 2 == 0, set1::add)
        .andThen(classify(n -> n % 3 == 0, set2::add))
        .andThen(classify(n -> n % 5 == 0, set3::add));

// Here the stream is consumed by the composed consumer
stream.forEach(multiClassifier);

Each inner consumer is created with the utility method defined above, which receives an independent predicate that, when matched, will add the element of the stream to the given set, i.e. if the element of the stream is a multiple of 3, it will be added to set2.

At the end, the stream is consumed with this composed consumer and thus the stream is classified by independent predicates:

System.out.println(set1); // [2, 4, 6, 8, 10, 12]
System.out.println(set2); // [3, 6, 9, 12]
System.out.println(set3); // [5, 10]