HTTP + JSON Pipeline

Source: 04_http_json_pipeline

Fetches a list of posts from the JSONPlaceholder public API, parses the JSON response, and writes the result to a local file. This example uses no stateful models; every operation is a pure dataflow connection scheduled by Mélodium as data becomes available.

Running


melodium run 04_http_json_pipeline/Compo.toml
# or with a custom output path:
melodium run 04_http_json_pipeline/Compo.toml --output results.txt

Expected output:


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

The file results.txt is written with the raw JSON body of the API response.

How it works

main starts two sub-treatments concurrently as soon as startup fires: logFetch logs a status message, and fetchPosts makes the HTTP GET request via get. Once the response body arrives as a Stream<byte>, it flows into 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 See in Compositeur Studio

parsePosts is a sub-treatment that chains four operations in a single connection statement:


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 converts the byte stream to UTF-8 text
toJson parses the text as JSON, emitting Option<Json>
unwrapOr<Json>(default=|null()) replaces a failed parse with null rather than aborting
toString<Json>() serialises the value back to a plain string for writing

Once writing finishes, writePosts.finished triggers logDone.

Dependencies


[dependencies]
std      = "0.10.1"  # core flows, logging, data structures
http     = "0.10.1"  # HTTP server and client
json     = "0.10.1"  # JSON parsing and serialisation
fs       = "0.10.1"  # local file I/O
encoding = "0.10.1"  # UTF-8 encode / decode