Java : Qu'est ce que l'API Gatherer ?

L'API Gatherer apparaît en preview à partir de java 22. Souvenez vous, on vous l'évoquait ici ! Elle est fortement liée aux streams. Découvrez dans cet article, ce qu'elle nous permet de faire !

⚠️Attention ! Cet article part du principe que vous connaissez déjà les bases du langage java, en particulier les notions autour des streams. Si ce n'est pas le cas, nous vous conseillons d'aller auparavant faire un tour sur les articles suivants :
Langage Java
L'API Stream

Un peu de contexte

Java 8 a été une véritable révolution dans le monde Java. L'API stream a fait son apparition, facilitant de loin les travaux des développeurs. Au cours des versions successives de java, elle a été améliorée, augmentée avec des nouvelles opérations.

Pour ne citer que quelques exemples :

Java 9 : takeWhile qui permet de parcourir le stream tant que les éléments consommés respectent la condition en paramètre
Java 9: ofNullable qui permet de retourner un stream contenant un seul élément s'il n'est pas null, ou un stream vide sinon
Java 16 : mapMulti qui permet de faire un mapping filtré sur le stream

Ces opérations sont dites "non terminales" car elles retournent un stream.

Il est possible, en java, d'implémenter ses propres opérations terminales (collect, reduce...), mais le besoin semble émerger d'implémenter ses propres opérations non terminales ! C'est exactement l'objectif de l'API Gatherer !

Alors comment ça marche ?

default <R> Stream<R> gather(Gatherer<? super T,?,R> gatherer)

Cette méthode de Stream permet d'appliquer une opération intermédiaire définie par l'objet "gatherer" qui est une instance de l'interface "Gatherer".

L'interface Gatherer

Elle est paramétrée par 3 types : <T, A, R>.
- T : Le type des éléments en entrée
- A : Le type de l'état mutable du Gatherer (oui, oui, on peut gérer un état mutable ! Ne vous inquiétez pas on va le voir ensemble 😊)
- R : Le type des éléments dans le stream de sortie
Elle contient 4 méthodes à connaître :
- Supplier<A> initializer()
- Integrator<A, T, R> integrator()
- BinaryOperator<A> combiner()
- BiConsumer<A, Downstream<? super R>> finisher()

Une fois ces méthodes maîtrisées, vous serez en mesure de créer votre propre opération intermédiaire !

Initializer

Objectif : Fournir la méthode d'initialisation de l'état mutable du Gatherer
Signature : Supplier<A> initializer();
- Le type A est le type de l'état du Gatherer

Integrator

Objectif : Fournir le cœur de notre opération intermédiaire, définir son fonctionnement
Signature : Integrator<A, T, R> integrator();
- Le type A est le type de l'état du Gatherer. En effet, c'est l'intégrateur qui mettra à jour cet état mutable si nécessaire
- Le type T est le type du stream d'entrée
- Le type R est le type du stream de retour

Integrator est une interface fonctionnelle. Sa méthode à implémenter est :

boolean integrate(A state, T element, Downstream<? super R> downstream);

state est l'état courant
element est l'élément courant du stream d'entrée. C'est sur cet élément qu'il faut faire l'opération de transformation souhaitée avant de le pousser dans le stream de sortie
downstream représente le stream de sortie. Son type est défini par l'interface fonctionnelle Downstream. Ses méthodes :
- boolean push(T element) : méthode abstraite qui permet de pousser un élément dans le downstream
- boolean isRejecting() : méthode qui renvoi "false" par défaut et permet de définir si le stream rejette les futurs éléments (ils ne seront pas poussés dans le stream de sortie)

Son type de retour est un booléen qui définit si les éléments suivants du stream doivent être consommées.

❓Mais ce booléen et la méthode "isRejecting()" semblent redondants… Non ?

"isRejecting" empêche le push, mais continue la consommation des éléments, alors que le booléen empêche directement sa consommation

Combiner

Dans le cas du parallélisme, le combiner permet de combiner deux états de notre opération intérmédiaire pour en produire un nouveau.

BinaryOperator<A> combiner()

Finisher

BiConsumer<A, Downstream<? super R>> finisher()

Cette fonction permet d'effectuer une action finale sur le stream. Par exemple, dans le cas du parallélisme, notre état mutable aura accumulé tous les états intermédiaires. Le finisher permet alors de traiter ce dernier état afin de pousser ce qu'il faut dans le downstream.

Des factories bien pratiques !

Pour les gatherer


static <T, R> Gatherer<T, Void, R> ofSequential(Integrator<Void, T, R> integrator);

Cette méthode permet de créer un gatherer séquentiel. Tout stream parallèle d'entrée deviendra alors séquentiel. Ce gatherer est stateless, ce qui explique le type "Void".

Il est également possible de créer un gatherer séquentiel qui gère son état mutable

static <T, A, R> Gatherer<T, A, R> ofSequential(Supplier<A> initializer, Integrator<A, T, R> integrator)

Si le séquentiel n'est pas un besoin, les méthodes statiques "of" permettent de créer un Gatherer :


static <T, R> Gatherer<T, Void, R> of(Integrator<Void, T, R> integrator);

static <T, R> Gatherer<T, Void, R> of(Integrator<Void, T, R> integrator, BiConsumer<Void, Downstream<? super R>> finisher);

static <T, A, R> Gatherer<T, A, R> of(Supplier<A> initializer, Integrator<A, T, R> integrator,BinaryOperator<A> combiner, BiConsumer<A, Downstream<? super R>> finisher);

Pour les integrator

Une méthode utilitaire à connaître est :

static <A, T, R> Greedy<A, T, R> ofGreedy(Greedy<A, T, R> greedy) {
    return greedy;
}

❓Encore une notion ? Mais qu'est-ce encore que Greedy ?

Greedy est une interface qui étend l'interface Integrator. Cette interface est une optimisation permettant de spécifier que l'integrator n'interrompra jamais un stream. L'utilisation de Greedy, dans ce cas, permet à la JVM d'optimiser le traitement.

Implémentons une opération de filtre premier

Objectif

Implémenter une opération non terminale qui ne retient que les nombres premiers d'un stream entrant.

Analyse

Notre opération n'interrompt pas le stream. On peut donc utiliser un Greedy integrator
On considère que l'ordre est important, on travaille donc en séquentiel et on utilisera la factory de gatherer "ofSequential"
On ne traite pas d'état mutable, on ne gèrera donc pas d'état initial

Implémentation

public static void main(String[] args) {
    Gatherer.Integrator<Void, Integer, Integer> integrator =
            Gatherer.Integrator.ofGreedy((_, element, downstream) -> {
        if (isPrime(element)) {
            return downstream.push(element);
        }
        return true;
    });
    var gatherer = Gatherer.ofSequential(integrator);
    List<Integer> sample = List.of(0,1,2,5,7,9,11,13,15,18,19);
    System.out.println(sample.stream().gather(gatherer).toList());
}

Conclusion

Nous avons vu que l'API Gatherer est une nouveauté permettant d'implémenter ses propres opérations non terminales sur les streams. Il en existe déjà plusieurs et elles couvrent la plupart des besoins. Seul l'avenir nous dira si elle présente une véritable utilité dans notre code java, ou s'il s'agit d'un nice-to-have répondant à des besoins spécifiques.