loops?

[static-moonlight]

I'm trying to build a stream processor which reads messages from a NATS subject, extract a file list from those messages, executes a processor for each file, and eventually creating a new message and send it back to another NATS subject.

Currently I'm struggling with the "for each file" part. Is there some kind of loop, I can use to go over this?

{
  "files": [
    {"path":"a","size":1},
    {"path":"b","size":2},
    {"path":"c","size":3}
  ]
}

This json was extracted from the message using the mapper processor, meaning I can adjust this if necessary.

This is my current configuration:

input:
  nats_jetstream:
    urls: [nats://${NATS_USERNAME}:${NATS_PASSWORD}@${NATS_ENDPOINT}]
    subject: ${NATS_INPUT_SUBJECT}
    stream: ${NATS_INPUT_STREAM}
    deliver: ${NATS_INPUT_DELIVER}
    durable: ${NATS_INPUT_CONSUMER_NAME}

pipeline:
  processors:
    - mapping: |
        root.files = this.package.files
    # for each file???

output:
  nats_jetstream:
    urls: [nats://${NATS_USERNAME}:${NATS_PASSWORD}@${NATS_ENDPOINT}]
    subject: ${NATS_OUTPUT_SUBJECT}

I could need a little push into the right direction :)

[rockwotj]

You can use the fold function in bloblang https://docs.redpanda.com/redpanda-connect/guides/bloblang/methods/#fold

Or the while processor: https://docs.redpanda.com/redpanda-connect/components/processors/while/

[rockwotj]

Here is an example I made a while back on looping over each element inside a message: https://gist.github.com/rockwotj/1a8b17eb8733743e1e0fafd32242c270

[static-moonlight]

oh my ... that looks a little complicated ^^

[static-moonlight]

I was hoping that for_each would automatically iterate through the json array. Currently I have this right before the for_each:

[
  {"path":"a","size":1},
  {"path":"b","size":2},
  {"path":"c","size":3}
]

    - mapping: |
        root = this.package.files
    - for_each:
      - log:
          level: INFO
          message: file
          fields_mapping: |
            root.files = this

I was expecting to see 1 log message for each file, but I only have 1 iteration, and that for the entire json structure.

Do I need to do something to split it somehow?

EDIT: The documentation speaks of batches?! At least if I understand it correctly ... So how do I split the json into multiple batches?

[rockwotj]

Yeah for_each doesn't do what you want, you might just want unarchive with the json_array format: https://docs.redpanda.com/redpanda-connect/components/processors/unarchive/

[static-moonlight]

This works like a charm, thanks :)

pipeline:
  processors:
    - mapping: |
        root = this.package.files.map_each(f -> f.path)
    - unarchive:
        format: json_array
    - for_each:
      - log:
          level: INFO
          message: transfer file
          fields_mapping: |
            root.path = this