Lançons le code précédent avec la commande sbt run et voyons le résultat de l’exécution :

[info] Done packaging.
[info] Running com.soat.techclub.MainExecutor
[akka.actor.default-dispatcher-2] INFO c.s.t.MainExecutor$ - Tripled => 111
[akka.actor.default-dispatcher-2] INFO c.s.t.MainExecutor$ - Tripled => 222
[akka.actor.default-dispatcher-2] INFO c.s.t.MainExecutor$ - Tripled => 333
[akka.actor.default-dispatcher-2] INFO c.s.t.MainExecutor$ - Tripled => 444
...

On voit bien comment chaque entier est dupliqué 3 fois puis affiché dans le message de log.

Dans cet exemple, il existe certes des détails d’implémentation spécifiques à Akka Streams comme la terminologie Source, Flow, Sink et la méthode de construction du flux. Ou bien aussi des éléments syntaxiques et sémantiques propres au langage Scala.
Mais en coulisses, c’est toujours la même architecture définie précédemment qui régit le fonctionnement du système réactif.
L’empreinte du modèle d’un flux ressemblant à une chaîne de montage industrielle reste fortement ressentie.

• Notes générales surScala
  1. Les mots clés val et var sont utilisés respectivement pour la déclaration des constantes et des variables.
     Par exemple, la syntaxe de déclaration d’une variable nommée xyz de type T est <strong>var</strong> xyz : T = ...
  2. Placé devant une val/var, une méthode ou une classe, le mot clé implicit rendra disponible cette val/var, méthode ou classe discrètement ou encore implicitement dans le contexte ou scope actuel.
     Concrètement, prenons le cas d’une méthode qui dans sa signature déclare un paramètre comme étant implicit. Il faudra lors de l’invocation soit lui passer explicitement un argument soit le déclarer implicit dans le scope en cours sans le lui passer.
     Dans l’exemple précédent, le constructeur ActorMaterializer() a besoin d’un objet ActorSystem implicite dans le contexte pour créer son instance. C’est pour cela que la constante val actrSys a été marquée comme implicit.

Modularité et composition

La section précédente a introduit les 3 composants fondamentaux de Akka Streams à savoir Source, Flow et Sink qui lorsque connectés forment le flux ou graphe linéaire de base RunnableGraph. Mais qu’arriverait-il si on connectait ces étages que partiellement ? Que pourrait-on bien avoir ?
Faisons l’expérience avec le snippet suivant :

val source = Source (1 to 10)
val flow = Flow[Int].map( i => i.toString )
val stage = source.via(flow)
logger.info(s"Class is >>> ${stage.getClass}")

On instancie un étage Source qui émet une séquence de Int de 1 à 10, puis un étage Flow qui transforme les Int en String. Par la suite on connecte les deux étages source et flow grâce à la méthode via et on récupère le résultat dans la constante stage.
Affichons la classe de stage avec le formatage de la chaîne de caractère préfixée par s.

• Notes générales sur Scala
  1. Pour formater une chaîne de caractère dans Scala il faut la préfixer avec s.
     L’interprétation du contenu se fait avec le symbole $ pour une valeur simple et avec ${ …​ } pour une expression.
     Cela s’appelle String interpolation.

Nous obtenons à la console :

[run-main-2] INFO c.m.l.MainExecutor$ - Class is >>> class akka.stream.scaladsl.Source 

La classe obtenue est de type Source.

On conclut donc qu’un objet Source[T,_] raccordé à un objet Flow[T,G,_] donne lieu à un objet de type Source[G,_] (l’emploi du _ signifie qu’on ignore le type de la matérialisation).
Et quand on enchaîne avec le même raisonnement, si on reprend le dernier objet de type Source[G,_] et qu’on le raccorde à un autre objet de type Flow[G,P,_] cela donne lieu à un objet Source[P,_].
La conclusion finale est que de façon globale, si on raccorde Source[T0,_] avec Flow[T0,T1,_] puis Flow[T1,T2,_] jusqu’à Flow[Tn-1,Tn,_] on obtient un objet dont le type est Source[Tn,_].

Traduisons cela en code :

val source = Source (1 to 10) 

val flow0 = Flow[Int].map( i => i.toFloat ) 
val flow1 = Flow[Float].map(f => f.toString) 
val flow2 = Flow[String].map(str => str.length < 4 ) 

val stage: Source[Boolean,_] = source.via(flow0).via(flow1).via(flow2)

• Ligne 1
  une source qui génère des Int de 1 à 10

• Ligne 3
  flow0 reçoit des objets Int et retourne des Float

• Ligne 4
  flow1 reçoit des objets Float et retourne des String

• Ligne 5
  flow2 reçoit des objets String et retourne des Boolean

• Ligne 7
  On enchaîne les raccordements avec la méthode via pour former une Source[Boolean,_].
  À noter que si on change le type paramètre de stage à autre chose que Boolean le compilateur marquera le type déclaré en erreur. Cela est dû au fait que le dernier étage flow2 émet un Boolean donc la source résultante doit absolument retourner un objet Source[Boolean,_].

En appliquant symétriquement le même raisonnement aux objets de types Flow et Sink on arrivera aux conclusions suivantes :

• Le raccord de Flow[T0,T1,_] puis Flow[T1,T2,_] jusqu’à Flow[Tn-1,Tn,_] donnera lieu à un objet de type Flow[T0,Tn,_].
• Le raccord de Flow[T0,T1,_] puis Flow[T1,T2,_] jusqu’à Flow[Tn-1,Tn,_] avec en dernier un Sink[Tn,_] donnera lieu à un objet de type Sink[T0,_].

Construisons un exemple plus illustratif et plus concret qui aidera à mieux assimiler tous ces symboles.
Prenons le bout de code suivant :

def squareSource(intSeq:Seq[Int]) : Source[Int,_] = {
 
    Source[Int](intSeq)
               .via( Flow[Int].map(i => i * i) ) 
}

def formatSink(msg:String) : Sink[Int,_] = {
 
    Flow[Int]
        .map(i => msg.format(i))
        .via( 
             Flow[String].map(str => str concat " !!"))  
        .to( 
            Sink.foreach[String](str => logger.info(str)))  
}

squareSource(3 to 9).to(formatSink("Squared = %s")).run

• Ligne 1
  On déclare une fonction squareSource qui prend en paramètre une Seq de type Int (liste ou séquence d’entier) et qui retourne comme résultat une Source émettant des Int.

• Lignes 3,4
  À l’intérieur on crée un objet Source à partir de la liste passée en paramètre qu’on raccorde avec via à un Flow nouvellement créé. Cet objet Flow va émettre la valeur du carré de l’entier reçu comme le décrit l’expression lambda</codecode class="language-plaintext">i ⇒ i * i.
  Cette composition forme l’objet à retourner par la fonction.

• Ligne 7
  On déclare une fonction formatSink acceptant un paramètre de type String et retournant un Sink ayant en réception un type Int. Le paramètre contiendra un message à formater avec les objets Int reçus.

• Lignes 9,10,11,12,13,14
  on instancie et raccorde deux objets de type Flow. Le premier reçoit des Int dont il utilisera la valeur pour formater le paramètre msg et le deuxième concatènera à la chaîne formatée la chaîne !!. Un dernier objet Sink vient se raccorder avec to aux étages précédents pour afficher avec le logger le résultat obtenu.
  La composition de ces 3 étages forme l’objet Sink[Int] à retourner par la méthode.

• Ligne 17
  En une seule étape, on appelle les fonctions déclarées précédemment qui retourneront les étages construits puis on les raccorde en utilisant to uniquement (on dispose que d’un Source et un Sink) et enfin on exécute le graphe linéaire avec run.

En lançant notre petit programme toujours avec la commande sbt run on obtient :

[info] Running com.soat.techclub.MainExecutor
...
[akka.actor.default-dispatcher-4] INFO c.m.l.MainExecutor$ - Squared = 36 !!
[akka.actor.default-dispatcher-4] INFO c.m.l.MainExecutor$ - Squared = 49 !!
[akka.actor.default-dispatcher-4] INFO c.m.l.MainExecutor$ - Squared = 64 !!
[akka.actor.default-dispatcher-4] INFO c.m.l.MainExecutor$ - Squared = 81 !!

Le résultat obtenu ne diffère pas de ce qu’on aurait eu s’il y avait déclaration et raccord des étages unité par unité (sans recours au regroupement par modules).

L’important avantage apporté par cette technique est qu’il est désormais possible de regrouper plusieurs traitements répartis sur un nombre illimité de sous étages puis d’exposer l’étage résultant sous la forme d’une entité Source, Flow ou Sink.
Il suffira par la suite de raccorder le tout avec les bons appels méthodes (via et to) sans exposer le comportement intrinsèque de l’étage composé.
De plus, grâce au Materializer un étage composé ne sera réellement concrétisé que lorsqu’on appelle effectivement la méthode run.

Donc un étage, simple ou composé, représente uniquement une sorte de plan descriptif du mode de fonctionnement et non pas le fonctionnement en soi.
C’est-à-dire qu’à chaque fois qu’on effectue une matérialisation concrète, on obtient une exécution nouvelle non liée aux autres exécutions émanant du même étage.
En conséquence de tout cela, on pourra exporter les étages composés vers d’autres utilisateurs de façon complètement transparente les rendant de la sorte portables et partageables.

Récapitulons l’essentiel de la composition avec l’illustration suivante :

• Notes générales sur Scala
  1. La syntaxe de déclaration d’une méthode ou fonction nommée fooFunc acceptant une liste de paramètre a de type A, b de type B, c de type C …​etc et retournant un résultat de type R est comme suit :
     <strong>def</strong> fooFunc <strong>(</strong>a<strong>:</strong>A<strong>,</strong> b<strong>:</strong>B<strong>,</strong> c<strong>:</strong>C<strong>)</strongstrong>:</strong> R <strong>=</strongstrong>{</strong> ... <strong>}</strong>
     Il est possible d’omettre l’expression du type de retour : R étant donné que le compilateur Scala dispose aussi pour les méthodes et fonctions d’un mécanisme d’inférence de type.
  2. Dans le corps de la méthode/fonction il est possible d’omettre le mot clé return car le compilateur Scala retourne par défaut le résultat de la dernière expression.
     C’est pour cette raison que lors du dernier code snippet on n’a pas utilisé d’expression avec return puisqu’on retourne directement l’objet construit.
  3. Lorsque la méthode/fonction déclarée ne contient qu’une seule expression on peut omettre les accolades { et } délimitant son corps.
     Appliquées à la fonction squareSource de l’exemple précédent, ces règles conduisent à cette réécriture de la fonction (sans type de retour, sans accolades) :
     <strong>def</strong> squareSource<strong>(</strong>intSeq : Seq[Int]) <strong>=</strong> Source[Int](intSeq).via(Flow[Int].map(i => i * i))

Récapitulatif partie 2

Voilà ! 😃
On arrive à la fin de ce deuxième blog post de la série dans lequel on avait premièrement défini la structure de notre projet, puis présenté le modèle système d’acteur et enfin exploré l’architecture de base de Akka Streams et sa propriété de modularité et composition.

Dans la troisième et dernière partie de notre série, on va se rapprocher du concret avec un cas d’utilisation plus tangible : une manipulation sur du contenu d’un fichier.

© SOAT

Toute reproduction interdite sans autorisation de la société SOAT