Generic Benthos HTTP Server-Sent-Events ingest and parser

Benthos HTTP-SSE parser

input:
  broker:
    copies: 1
    inputs:
      - http_client:
          url: "https://mas.to/api/v1/streaming/public"
          stream:
            enabled: true
            reconnect: true
            codec: "delim:\n\n"

pipeline:
  processors:
    - unarchive:
        format: lines
    - mapping: |
        root = if content().string().has_prefix(":") { deleted() } else { content().string().parse_yaml() }
    - archive:
        format: json_array
    - mapping: |
        root = this.squash().with("id","data","event","retry")
        root.data = if root.exists("data") && root.data.type() != "array" { [root.data] }