Transformer des flux

Akka Streams fournit un riche ensemble d'opérateurs sur les Sources et les Flows, qui rappellent l'API des collections de Scala, mais s'exécutent de manière asynchrone et respectent la contre-pression.

Chaque opérateur retourne un nouveau modèle ; les transformations sont donc composées de manière déclarative avant même l'exécution du flux.

map applique une fonction synchrone à chaque élément ; filter supprime les éléments qui ne satisfont pas un prédicat. Ce sont les outils essentiels des transformations élément par élément.

Tous deux préservent l'ordre et propagent l'achèvement ainsi que les échecs vers l'aval.

Lorsqu'une entrée doit produire plusieurs sorties, utilisez mapConcat. Il prend une fonction qui renvoie un itérable et aplatit les résultats dans le flux.

Retourner une collection vide revient à supprimer l'élément.

grouped(n) regroupe les éléments consécutifs dans un Seq contenant au plus n éléments, ce qui est utile pour les écritures en masse dans une base de données. sliding(n) émet des fenêtres qui se chevauchent.

Le traitement par lots réduit le coût par élément dans les pipelines fortement dépendants des entrées-sorties.

scan émet l'accumulateur courant après chaque élément, ce qui produit un flux d'état évolutif. fold n'émet la valeur accumulée finale qu'une fois la source amont terminée.

Utilisez scan pour les compteurs en temps réel et fold pour les agrégats finaux.

mapAsync(parallelism) appelle une fonction qui renvoie un Future et émet les résultats dans l'ordre, en exécutant simultanément jusqu'à parallelism futures.

Utilisez-le pour les appels asynchrones, comme les recherches en base de données ou les requêtes HTTP, lorsque l'ordre est important.

mapAsyncUnordered se comporte comme mapAsync, mais émet chaque résultat dès qu'il est terminé, sans tenir compte de l'ordre des entrées.

Il peut améliorer le débit lorsque l'étape en aval ne dépend pas de l'ordre, car un traitement asynchrone lent ne bloque plus les traitements plus rapides.

Pour les transformations élément par élément qui nécessitent un état local mutable, statefulMapConcat crée un nouvel état à chaque matérialisation et renvoie un itérable de sorties.

C'est la manière sûre de conserver des compteurs ou des tampons sans partager l'état entre les exécutions du flux.

Les flux peuvent effectuer des transformations en fonction du temps aussi bien que du contenu. throttle plafonne le débit d'émission, groupedWithin regroupe les éléments selon une taille ou une durée écoulée, et takeWithin limite la durée.

Ces opérateurs sont essentiels pour limiter le débit des API externes.

Par défaut, une exception levée dans un opérateur fait échouer l'ensemble du flux. Une stratégie de supervision peut à la place resume (ignorer l'élément incorrect) ou restart l'étape.

Associez la stratégie au flux avec withAttributes.

De petits flux se composent en flux plus grands avec via, ce qui produit un flux unique et réutilisable. Chaque transformation reste ainsi ciblée et peut être testée indépendamment.

Le flux composé possède le type d'entrée du premier flux et le type de sortie du dernier.

Réfléchissez aux transformations asynchrones et à leurs garanties concernant l'ordre.

Vous avez étudié les opérateurs de transformation : map/filter élément par élément, mapConcat pour produire plusieurs éléments à partir d'un seul, grouped pour regrouper les éléments, scan et fold pour les accumuler, ainsi que mapAsync pour les traitements asynchrones.

Vous avez également découvert les transformations avec état, les opérateurs fondés sur le temps comme throttle, les stratégies de supervision des erreurs et la composition des flux avec via. Ensuite : comment la contre-pression sécurise ces étapes.