La 13ème édition du Devoxx France s’est tenue au Palais des Congrès à Paris du mercredi 16 au vendredi 18 avril 2023. À cette occasion, José Paumard a présenté une conférence intitulée "L'API Gatherer : l'outil qui manquait à vos Streams".

 1 – Les limitations des Streams

Pour rappel, un Stream est un pipeline de traitement des données, composé d’une source, d’opérations intermédiaires, et d’une opération terminale. L’ensemble fonctionne de manière fluide mais rigide : chaque Stream ne peut comporter qu’un seul pipeline unique, sans possibilité d’ajouter plusieurs opérations intermédiaires ou terminales.

Limitations des Streams

• Absence de personnalisation : Vous ne pouvez pas créer vos propres opérations intermédiaires.

• Rigidité structurelle : Les opérations terminales ne peuvent pas interrompre le flux en cours de manière personnalisée (à l’exception de méthodes comme findFirst() ou limit()).

• Conception inflexible : Toutes les opérations intermédiaires doivent s’inscrire dans les types standard fournis par Java, sans possibilité d’adapter les comportements.

2 – API Gatherer : une révolution dans les Streams

L’API Gatherer comble ces lacunes en permettant la conception d’opérations intermédiaires sur mesure.

Elle fournit trois composants fondamentaux, détaillés dans le document :

2.1 - Interface Gatherer

L’API Gatherer nous donne 3 éléments.

• Le premier élément c’est une interface qui s’appelle Gatherer, paramétré par 3 types T, A, et R

Interface Gatherer<T,A,R>

  Du coup ça ressemble beaucoup aux trois types de l’interface Collector Interface Collector <T,A,R>

• Le second élément, c’est une méthode intermédiaire sur API Stream qui s’appelle gather(), et qui prend un Gatherer en paramètre.

• Enfin on a une classe factory Gatherers, qui est le pendant de la classe Factory Collectors, à la différence qu’elle est beaucoup plus simple, puisqu’elle n’a que cinq méthodes, contre une cinquantaine pour Collectors.

Différence entre Gatherer et Collector

Si on compare l’interface Gatherer et Collector, on constate que le type T, c’est le type des éléments que le Gatherer va consommer.

Supposons que j’ai un upstream qui produit des éléments de type T qui les pousse vers ce gatherer

et le type R est le type des éléments qui va pousser vers le downstream, si je compare avec l’API Collector, le R n’a pas exactement le même sens.

Ce qui peut prêter à confusion, c’est que dans le contexte de Collector, R désigne la structure finale contenant tous les éléments, alors qu’avec Gatherer, R est le type des éléments individuels poussés à l’étape suivante. Cela veut dire qu’il retourne un Stream de R.

Le type A dans l’interface Gatherer<T, A, R> représente l’état intermédiaire utilisé pour l’agrégation ou la transformation des éléments de type T avant qu’ils ne deviennent des R poussés vers le downstream.

Dans l’API Collector, A est souvent une structure mutable qui stocke temporairement les éléments avant de produire R. Par exemple, A pourrait être une List<T> qui accumule des éléments avant de les convertir en une Set<T> ou une Map<T, U>.

Dans Gatherer, c’est un peu différent : A peut-être un état mutable ou une logique interne qui dicte comment chaque élément T est transformé avant d’être transmis en tant que R. Il ne s’agit pas forcément d’un conteneur, mais plutôt d’un mécanisme de transformation qui façonne le flux de données.

Une façon de voir les choses :

• Avec Collector, A est une boîte d’accumulation.

• Avec Gatherer, A est une boîte de traitement.

2.2       Integrator

2.2.1   Lazy Integrator

Dans l'API Gatherer de Java 22, l'Integrator par défaut est généralement lazy, ce qui signifie qu'il n'intègre les éléments qu'au fur et à mesure et ne consomme pas l'ensemble du flux immédiatement

Cet Integrator est une interface fonctionnelle qui prend 3 éléments :

• State

• Element

• downstream

• … et qui retourne un boolean.

N.B :

Cet Integrator est paramétré par A, T et R, alors que le Gatherer est paramétré par T, A et R. Donc il faut faire attention quand on construit un Integrator de ne pas se tromper dans l’ordre des paramètres.

L’Integrator est l’objet qui a la responsabilité d’intégrer les éléments que l’upstram de notre Gatherer va pousser vers le Gatherer (en fait ces éléments vont tomber dans l’intégrateur).

Et l’Integrator a aussi la responsabilité d’intégrer ces éléments dans le downstream modélisé par cet objet. 

Si on veut modéliser un Gatherer, on peut l’écrire en utilisant une méthode factory de l’interface Gatherer qui s’appelle of.

Celle-ci nous permet de passer l’integrator directement en paramètre

Ici, l’Integrator est :

Et il joue exactement le même rôle que l’accumulator dans un Collector.

Ici notre Gatherer émet la somme cumulative. Ci-dessous un exemple d’utilisation de ce Gatherer

2.2.2   Greedy Integrator

 Si jamais on a un Integrator qui ne fait que retourner la valeur qui a été transmise par le downstream.push(), et donc qui ne choisit pas de retourner faux de lui-même, qui va, soit systématiquement retourner vrai, soit retourner la valeur retournée par le push (ce qui est le cas du mappage ou du filtrage), alors notre Integrator devient un Greedy integrator dans le vocabulaire de l’API Gatherer.

Et un operateur greedy, c’est juste un operateur que vous avez créé avec la méthode ofGreedy de l’interface Integrator.

Lorsque vous déclarez votre intégrateur comme Greedy, l’API peut alors optimiser son comportement. Cela s’appuie notamment sur le fonctionnement de la méthode Splitterator, qui propose deux méthodes principales :

• tryAdvance : traite un élément à la fois.

• forEachRemaining : traite tous les éléments d’un coup sans faire de test — elle "pousse tout sans réfléchir".

Dans le cas d’un intégrateur Greedy, cette deuxième méthode peut être utilisée de manière optimisée. Ces optimisations ne sont possibles que si l’opérateur est déclaré Greedy.

Par défaut, vos opérateurs ne sont pas Greedy — vous n’avez probablement jamais écrit explicitement integrator.of ou équivalent dans votre code. Cela signifie que :

• Vous commencez à écrire votre opérateur sans vous soucier de Greedy.

• À la fin, vous évaluez s’il est effectivement Greedy ou non.

• Si vous ne savez pas, vous ne changez rien.

• Si vous savez qu’il est Greedy, alors vous le déclarez explicitement comme tel.

2.3 - Downstream

2.3.1   Push method

Vous remarquerez que pour passer un élément à notre downstream, il y a une méthode push sur cette interface, qui va pousser l’élément vers le downstream suivant.

Cette méthode push retourne un booléen.

Dans notre exemple ci-dessus on ne s’en n’est pas servi.

Pourtant on met le doigt sur une fonctionnalité puissante et subtile de l’API Gatherer car ça a un effet direct sur le flux Stream.

Car elle permet à l’opération suivante de dire :

• true → “OK, continue à m’envoyer des éléments”

• false → “Stop, j’ai ce qu’il me faut, on arrête là.”

C’est une forme d’arrêt anticipé (short-circuiting), comme dans findFirst(), anyMatch(), etc…

Que se passe-t-il si on pousse des éléments vers un downstream qui n’en accepte plus ? Rien.

L’application ne crashera pas. Aucune exception ne sera levée.

Pour compte on sera en train de faire des calculs pour rien car les éléments seront jetés.

Maintenant, à quelle condition un downstream peut arrêter d’accepter des éléments?

On peut utiliser la fonction limit() à cet effet

Exemple :

On va créer un Gatherer tout simple, qui :

• multiplie chaque élément par 10

• pousse le résultat vers le downstream

• arrête de traiter dès que le downstream ne veut plus de données

Résultat

10

20

30

Et on n’ira pas plus loin, parce que limit(3) dit "stop" au 4e push.

Le Gatherer le détecte via le retour false de downstream.push(...).

 2.3.2   Le comportement de l’objet downstream

 Le downstream est un objet qui transporte un état, notamment l’état de rejet (Rejecting).

On a donc une méthode isRejecting() sur ce downstream qui vous permet de savoir s’il accepte les valeurs suivantes ou pas.

Ce downstream agit comme une porte :

• Ouverte : l’état initial (Rejecting == false).

• Fermée : après un changement d’état (Rejecting == true).

Il y a plusieurs règles avec le fonctionnement de cette porte qui ont un impact sur la façon dont on écrit les Integrator :

1. Elle démarre toujours ouverte.
   Lorsqu’un Integrator est appelé pour la première fois avec un downstream tout neuf, il est ouvert.

2. Elle commute uniquement de « ouvert » vers « fermé ».
   Une fois fermée, elle ne peut jamais se rouvrir.

3. Cet état ne peut switcher qui si on fait un push.
   ça veut dire que :

   • La porte ne se ferme pas d’elle-même.

   • À chaque push, elle peut rester ouverte ou se fermer.

   • Pas de fermeture spontanée sans push.

 Il est possible de créer vos propres implémentations de downstream, qui ne suivent pas ces règles.
Par exemple :

• Une implémentation qui commute en fonction d’une horloge, et non d’un push.

Cela peut être utile dans certains cas, mais vous devez en être conscient.
Le JDK ne fournit pas d’implémentation downstream alternative de ce type — c’est à vous de gérer cela si vous le faites.

Maintenant prenons le code suivant

Est-ce une bonne idée d’écrire ce code ?

A priori, ça ressemble à une optimisation : si le downstream est fermé, inutile de pousser des données. Mais est-ce vraiment une optimisation utile ? Non

À la première invocation, votre downstream est neuf, donc la porte est ouverte : isRejecting() retourne false.

Ensuite, elle ne peut commuter vers « fermé » que via un push() :

• Si push() retourne true, la porte reste ouverte.

• Si push() retourne false, la porte se ferme.

• Si vous retournez false dans votre Integrator, il n'est plus appelé.

Par conséquent le test ne sert à rien dans ce contexte, car isRejecting() est toujours false à ce moment-là.

Donc, ne le mettez pas.

Cela rend le code plus simple et plus clair.

3 – Conclusion

L’API Gatherer marque une avancée significative dans la manipulation des flux de données en Java, offrant une flexibilité jusqu’ici inégalée dans la gestion des opérations intermédiaires des Streams.

Alors que les Collectors ont longtemps été la solution privilégiée pour les opérations terminales, Gatherer vient combler un vide en permettant la création de traitements intermédiaires personnalisés.

L’API Gatherer est riche et complexe, et pour bien la maîtriser, il est essentiel de l’expérimenter en conditions réelles.

Pour aller plus loin, rien de mieux qu’une session de live coding. Je vous invite vivement à découvrir celui de José Paumard (New Livestream – Gatherers: The API Your Stream Was Missing), où il explore l’API Gatherer en profondeur à travers des démonstrations pratiques et des explications détaillées. Une excellente opportunité pour comprendre la puissance de cette nouvelle API et l’intégrer efficacement dans vos développements !

Annexe : Bibliographie de José Paumard José Paumard est un expert reconnu dans l'écosystème Java, combinant une carrière académique et industrielle. Actuellement Developer Advocate Java chez Oracle, il contribue activement à la documentation officielle sur dev.java et à la promotion des nouvelles fonctionnalités du langage.

Contributions notables :

• JEP Café : Une série mensuelle de vidéos sur YouTube, en anglais, explorant les nouveautés du JDK.

• Cracking the Java Coding Interview : Des vidéos courtes répondant à des questions fréquemment posées lors des entretiens d'embauche pour les développeurs Java.