Learning Clojure core.async

README.md

Mon apprentissage de core.async

J'ai décidé de pencher un peu sérieusement sur core.async. L'objectif de ce gist est d'accompagner mon apprentissage et ma compréhension.

core.async, c'est LA librairie de gestion de l'asynchrone en Clojure. Comme dans le langage Go, elle s'appuie sur le concept des CSP (Communicating Sequential Processes).

En matière d'asynchrone, je me suis souvent appuyé sur RxJS et les observables. Je dois avouer que ce passage à un autre paradigme n'est pas facile.

Les bases

Pour saisir les bases de la librairie et de sa syntaxe, les talks de Rich Hickey ici et Timothy Baldrige là ainsi que le walkthrough officiel sont de bons points de départ et je ne chercherais pas à faire mieux.

Mon problème se situe plus dans la structure du code dans un "vrai" programme.

Patterns

Une bonne référence en matière d'usage réel de la librairie est encore un autre talk de Timothy Baldrige. Dans ce talk, on apprend principalement deux choses : il faut utiliser des transducers et il est préférable de passer le channel à alimenter à la fonction qui produit les valeurs plutôt que de le créer dans la fonction.

Ce dernier point est essentiel, mais n'a pas été évident à saisir pour moi. En effet, dans le monde des observables, on appelle une fonction qui retourne l'observable, puis on map/filter/merge dessus pour la fournir à un consommateur.

Le premier pattern qui m'a permis de mieux appréhender le concept me vient de ce bout de code de Stuart Halloway.

Producer Consumer V1

Le fichier producer-consumer-1.clj en est une version simplifiée.

La fonction produce-values prend un channel en paramètre et écrit des valeurs dedans.

La fonction consume-values prend un channel en paramètre et lit celui-ci en boucle (et affiche la valeur), jusqu'à obtenir une valeur nulle (ce qui signifie que le channel est fermé).

A la ligne 15, on trouve un let qui fait la liaison entre ces deux fonctions : on crée un channel, on le donne à produce-values puis à consume-values.

Le code pourrait s'arrêter là et il serait parfaitement fonctionnel. Que font les lignes "supplémentaires" ?

Nos deux fonctions manipulent un channel, mais elles en retournent aussi un autre. Ceux-ci permettent de savoir quand le traitement est terminé.

Pour récupérer les valeurs émises (ou le signal de fermeture), on utilise <! qui doit donc se trouver dans un bloc go, celui retournant lui aussi un channel, on l'appelle avec <!! pour bloquer le traitement jusqu'à terminaison du traitement.

La valeur retournée est { :pdr nil :csr nil}, ce qui pour l'instant ne présente pas beaucoup d'intérêt. Nous verrons dans la V2 par quoi l'on peut remplacer ces valeurs nulles.

Producer Consumer V2

Dans les prochaines versions, nous nous concentrerons sur consume-values, mais les mêmes principes sont applicables à produce-values.

Un première usage que l'on peut faire du channel retourné par consume-values est un rapport du traitement. Le fichier producer-consumer-2.clj montre comment l'on peut indiquer par exemple le nombre de valeurs reçues.

La valeur retournée par le bloc est donc maintenant { :pdr nil :csr { :count 10 } }.

producer-consumer-1.clj

(require '[clojure.core.async :as a :refer (<! <!! chan go)])

(defn produce-values [ch]
  (a/onto-chan ch (range 10)))

(defn consume-values [ch]
  (go
    (loop []
      (if-let [v (<! ch)]
        (do 
          (println v)
          (recur))))))

(<!! (go
       (let [c (chan)
             pdr (produce-values c)
             csr (consume-values c)]
         {:pdr (<! pdr)
          :csr (<! csr)})))

producer-consumer-2.clj

(require '[clojure.core.async :as a :refer (<! <!! chan go)])

(defn produce-values [ch]
  (a/onto-chan ch (range 10)))

(defn consume-values [ch]
  (go
    (loop [n 0]
      (if-let [v (<! ch)]
        (do
          (println v)
          (recur (inc n)))
        { :count n }))))

(<!! (go
       (let [c (chan)
             pdr (produce-values c)
             csr (consume-values c)]
         {:pdr (<! pdr)
          :csr (<! csr)})))

producer-consumer-3.clj

(require '[clojure.core.async :refer (<! <!! chan go)])

(defn produce-values [ch]
  (a/onto-chan ch (range 10)))

(defn consume-values [ch]
  (go
    (try
      (loop [n 0]
        (if-let [v (<! ch)]
          (do
            (println v)
            (recur (inc n)))
          { :count n }))
      (catch Throwable t
        { :error true :message (.getMessage t) }))))

(<!! (go
       (let [c (chan)
             pdr (produce-values c)
             csr (consume-values c)]
         {:pdr (<! pdr)
          :csr (<! csr)})))