Pipeline HTTP + JSON

Source: 04_http_json_pipeline

Récupère une liste de posts depuis l’API publique JSONPlaceholder, parse la réponse JSON et écrit le résultat dans un fichier local. Cet exemple n’utilise aucun modèle avec état ; chaque opération est une connexion de dataflow pur ordonnancée par Mélodium au fur et à mesure que les données sont disponibles.

Exécution


melodium run 04_http_json_pipeline/Compo.toml
# ou avec un chemin de sortie personnalisé :
melodium run 04_http_json_pipeline/Compo.toml --output results.txt

Sortie attendue :


[…] info: pipeline: fetching posts...
[…] info: pipeline: done

Le fichier results.txt est écrit avec le corps JSON brut de la réponse API.

Fonctionnement

main démarre deux sous-traitements en parallèle dès que startup se déclenche : logFetch journalise un message de statut, et fetchPosts effectue la requête HTTP GET via get. Une fois le corps de la réponse disponible en tant que Stream<byte>, il s’écoule dans parsePosts :


treatment main(const output: string = "results.txt")
{
    startup()
 
    fetchPosts: get(url="https://jsonplaceholder.typicode.com/posts")
    logFetch:   logInfoMessage(label="pipeline", message="fetching posts...")
 
    startup.trigger -> logFetch.trigger
    startup.trigger -> fetchPosts.trigger
 
    parsePosts()
    writePosts: writeTextLocal(path=output, append=false)
    logDone:    logInfoMessage(label="pipeline", message="done")
 
    fetchFailed: logErrorMessage(label="http", message="fetch failed")
    fetchError:  logError(label="http")
 
    fetchPosts.data    -> parsePosts.data,text -> writePosts.text
    fetchPosts.failed  -> fetchFailed.trigger
    fetchPosts.error   -> fetchError.message
    writePosts.finished -> logDone.trigger
}

main treatment diagram Voir dans Compositeur Studio

parsePosts est un sous-traitement qui enchaîne quatre opérations en une seule instruction de connexion :


treatment parsePosts()
  input  data: Stream<byte>
  output text: Stream<string>
{
    decode()
    toJson()
    unwrapOr<Json>(default=|null())
    toString<Json>()
 
    Self.data -> decode.data,text -> toJson.text,json -> unwrapOr.option,value -> toString.value,into -> Self.text
}

decode convertit le flux d’octets en texte UTF-8
toJson parse le texte en JSON, émettant Option<Json>
unwrapOr<Json>(default=|null()) remplace un échec de parsing par null plutôt que d’interrompre
toString<Json>() sérialise la valeur en chaîne simple pour l’écriture

Une fois l’écriture terminée, writePosts.finished déclenche logDone.

Dépendances


[dependencies]
std      = "0.10.1"  # flux de base, journalisation, structures de données
http     = "0.10.1"  # serveur et client HTTP
json     = "0.10.1"  # parsing et sérialisation JSON
fs       = "0.10.1"  # lecture/écriture de fichiers locaux
encoding = "0.10.1"  # encodage / décodage UTF-8