This Benthos config demonstrates the consumption of Mastodon HTTP Server-Sent Events

Benthos Mastodon streaming input

input:
  broker:
    copies: 1
    inputs:
      - http_client:
          url: "https://mas.to/api/v1/streaming/public"
          stream:
            enabled: true
            reconnect: true
            codec: lines
            max_buffer: 100000000
        processors:
          - mapping: |
                root = if content().string().has_prefix(":") { deleted() }
    batching:
      count: 2

pipeline:
  processors:
    - grok:
        expressions:
          - '%{HTTPSSE}'
        pattern_definitions:
          HTTPSSE: "%{NOTSPACE:key}: %{GREEDYDATA:value}"
    - archive:
        format: json_array
    - mapping: |
        root.event = this."0".value
        root.data = this."1".value.parse_json()
output:
  stdout:
    codec: lines