$ whoami

Piotr Przybył
piotrprz
@piotrprz@mstdn.social
piotrprz.bsky.social
Senior Developer Advocate

Java Champion
Testcontainers Community Champion

$ who are you

CAVEAT AVDITORES!

A.K.A. Safe harbour statement: don't assume anything based on this presentation. Verify on your own. Errare humanum est.

Agenda

Prologue

4025 days ago

Java 8 was released

Lambdas

Streams

a stream

• source (collection, spliterator, ...)
• intermediate operation(s)
• terminal operation

Intermediate operations

• filter
• map
• mapToInt
• mapToLong
• mapToDouble
• flatMap
• flatMapToInt
• flatMapToLong
• flatMapToDouble
• distinct
• sorted
• sorted(Comparator c)
• peek
• limit
• skip
• boxed
• sequential
• parallel
• unordered
• ...

Terminal operations

• forEach
• forEachOrdered
• toArray
• reduce
• collect
• min
• max
• count
• anyMatch
• allMatch
• noneMatch
• findFirst
• findAny
• ...

Intermediate operations,
continued

In Java 9:

• takeWhile
• dropWhile

In Java 16:

• mapMulti
• mapMultiToInt
• mapMultiToLong
• mapMultiToDouble

Terminal operations,
continued

In Java 16:

• toList

Terminal operations, continued for real!

In Java 9:

• Collectors.filtering
• Collectors.flatMapping

In Java 12:

• Collectors.teeing

In ecosystem:

• Parallel collectors

And in your own Utils!

Part one:

The Why

Or better said: the problems

    static Stream<String> eneStream() {
        return Stream.of("""
            Ene due rike fake \
            torba borba ósme smake \
            eus deus kosmateus \
            i morele baks"""
            .split(" "));
    }
            

How to get only one word by length?

        Set<String> uniqueLengthWords = eneStream()
            .collect(
                Collectors.collectingAndThen(
                    Collectors.toMap(
                        String::length,
                        word -> word,
                        (existing, _) -> existing
                    ),
                    map -> new HashSet<>(map.values())
                )
            );
            

Need to reuse?

    Collector<String, HashMap<Integer, String>, Set<String>> uniqueLengthCollector() {
        return Collector.of(
            HashMap::new,
            (map, word) -> map.putIfAbsent(word.length(), word),
            (map1, map2) -> {
                map2.forEach(map1::putIfAbsent);
                return map1;
            },
            map -> new HashSet<>(map.values())
        );
    }
            

What if we need to lower-case them all?

    private static void withCollectorAndUp() {
        Set<String> uniqueLengthWords = eneStream()
            .collect(Collectors.collectingAndThen(
                uniqueLengthCollector(),          // Apply the provided collector
                set -> set.stream()
                    .map(String::toUpperCase)
                    .collect(Collectors.toSet()) // Convert collected words to uppercase
            ));
    }
            

Not perfect, we create another stream...

or we could map() all elements...

or we could decorate all elements so that we can use Stream.distinct...

or we could have another pr0 and f4ncy solution...

or we could create a ticket "PLZ ADD Stream.distinctBy(...)"!

They could add distinctBy(...)!

And two dozen other methods which are needed by someone, the need is legit and justified.

Only then we have Stream API bloated even more by a promile of users...

Part two:

The What

Don't hallucinate, please ;-)

TL;DR:

            Stream.of(...)
                  .gather(a)
                 [.gather(b)]
                 [.gather(c)]
                  .collect(...);
        

            Stream.of(...).gather(a)[.andThen(b).andThen(c)].collect(...);
        

Stream gatherers can:

• transform elements 1:1, 1:N, N:1, M:N
• track already processed elements
• "suspend" elements
• short-circuit (∞ → | )
• keep state between executions
• enable parallel execution

Part three:

The How

Let's try again

        Set<String> uniqueLengthWords = eneStream()
            .gather(new UniqueByLengthGatherer())
            .map(String::toUpperCase)
            .collect(Collectors.toUnmodifiableSet());
        

class UniqueByLengthGatherer implements Gatherer<String, Set<Integer>, String> {
    public Supplier<Set<Integer>> initializer() {
        return HashSet::new;
    }

    public Integrator<Set<Integer>, String, String> integrator() {
        return (state, element, downstream) -> {
            if (!state.contains(element.length())) {
                state.add(element.length());
                downstream.push(element);
            }
            return true;
        };
    }
}
        

Let's try again

        Set<String> uniqueLengthWords = eneStream()
            .gather(Gatherer.ofSequential(HashSet::new, integrator))
            .map(String::toUpperCase)
            .collect(Collectors.toUnmodifiableSet());

        

        Gatherer.Integrator<Set<Integer>, String, String> integrator =
            (state, element, downstream) -> {
            if (!state.contains(element.length())) {
                state.add(element.length());
                downstream.push(element);
            }
            return true;
        };
        

Formal definition

            public interface Gatherer<T, A, R> {
                default Supplier<A> initializer() { }
                Integrator<A, T, R> integrator();
                default BinaryOperator<A> combiner() { }
                default BiConsumer<A, Downstream<? super R>> finisher() { }
            }
        

Formal definition - longer

            public interface Gatherer<IN, STATE, OUT> {
                default Supplier<STATE> initializer() { }
                Integrator<STATE, IN, OUT> integrator();
                default BinaryOperator<STATE> combiner() { }
                default BiConsumer<STATE, Downstream<? super OUT>> finisher() { }
            }
        

On site construction

Gatherer.ofSequential(integrator);

Gatherer.ofSequential(integrator, finisher);

Gatherer.ofSequential(initializer, integrator);

Gatherer.ofSequential(initializer, integrator, finisher);

Gatherer.of(integrator);

Gatherer.of(integrator, finisher);

Gatherer.of(initializer, integrator, combiner, finisher);

Integrator

boolean integrate(STATE currState, IN nextElem, Downstream<? super OUT> downstream);

• consumes nextElem
• may inspect and/or update currState
• can send any number of items to downstream
• return true → more integration

interface Greedy extends Integrator

• will never short-circuit

Combiner

combines two parallelizable gatherers

Finisher

can still operate on downstream

Downstream

can push things down the stream

Built-in gatherers

Gatherers.fold(initial, BiFunction folder);

Gatherers.scan(initial, BiFunction scanner);

Gatherers.windowFixed(size);

Gatherers.windowSliding(size);

Gatherers.mapConcurrent(maxConcurrency, Function mapper);

Built-in gatherers

    private void folding() {
        var list = IntStream.range(0, 10)
            .mapToObj(i -> "" + i)
            .gather(Gatherers.fold(() -> "", (string, number) -> string + number))
            .toList();
        System.out.println(list);
    }
    // [0123456789]



    private void scanning() {
        var list = IntStream.range(0, 10)
            .mapToObj(i -> "" + i)
            .gather(Gatherers.scan(() -> "", (string, number) -> string + number))
            .toList();
        System.out.println(list);
    }
    // [0, 01, 012, 0123, 01234, 012345, 0123456, 01234567, 012345678, 0123456789]



    private void fixed() {
        var list = IntStream.range(0, 10)
            .mapToObj(i -> "" + i)
            .gather(Gatherers.windowFixed(3))
            .toList();
        System.out.println(list);
    }
    // [[0, 1, 2], [3, 4, 5], [6, 7, 8], [9]]



    private void sliding() {
        var list = IntStream.range(0, 10)
            .mapToObj(i -> "" + i)
            .gather(Gatherers.windowSliding(3))
            .toList();
        System.out.println(list);
    }
    // [[0, 1, 2], [1, 2, 3], [2, 3, 4], [3, 4, 5], [4, 5, 6], [5, 6, 7],
    // [6, 7, 8], [7, 8, 9]]
        

Epilogue

If only we could use streams efficiently more often...

            IntStream.range(0, 100).parallel().map(id -> enriched(id)).toList();
        

            IntStream.range(100, 200).parallel().map(id -> enriched(id)).toList();
        

Gatherers.mapConcurrent(maxConcurrency, Function mapper);

softwaregarden.dev/en/posts/new-java/gatherers/concurrent/

            IntStream.range(0, 200)
            .gather(Gatherers.mapConcurrent(100, id -> enriched(id))
            .toList()
        

Always

Everyone

Feedback

How was it?

Slides