allow flows to execute a process after processing input messages

[reubenmiller]

Is your feature request related to a problem? Please describe.

When writing a flow, it is currently not possible to trigger a process after messages are processed within a flow. Currently the only way a process can be executed within a flow is to run the input.process which then involves the user having to stitch together two flows by passing some state in-between them (via the local MQTT broker, or potentially files).

An example of this is if a user wants to create a flow that could be used to upload files from the local filesystem to Cumulocity.

The user wishes to scan a given directory, say /data/inbox, and upload each of the *.csv files, and upload them using the tedge upload c8y command.

Currently, this would be fairly difficult to do, but a very naive approach would be to create two flows where;

• Flow 1 would list the files within the /data/inbox folder, and write the output to
• Flow 2 would Read from the pre-configured location, and upload the files one by one and

Flow 1: List the files that should be uploaded

name = "file-uploader-part1"
version = "1.0.0"
description = "List files to be uploaded"

[input.process]
command = """find /data/inbox -name '*.csv'"""
interval = "60s"

steps = [
    { script = "main.js" },     # optional message filter
]

[output.file]
path = "/tmp/process.list"

Flow 2: Upload each file and send a notification

name = "file-uploader-part2"
version = "1.0.0"
description = "Upload files provided in the process.list file"

[input.process]
command = """tail -n 100 /tmp/process.list | xargs -n1 -I tedge upload c8y --file {}"""
interval = "60s"

steps = [
    { script = "main.js" }, # publish a local MQTT message to notify that the work was done
]

The above flows wouldn't work very well as the /tmp/process.list being used as a queue however it is easy to avoid processing messages twice if the flow is restarted.

Describe the solution you'd like

The following lists a few potential solutions (but don't limit the discover to just these listed solutions). Both solutions are common by not requiring the user to have to write two flows to solve the above problem, but allow a process to be executed at additional positions (not just via input.process).

Solution 2: Allow a process to be executed in the output (preferred)

name = "file-uploader"
version = "1.0.0"
description = "Upload all files within a folder"

[input.process]
# list log files and print their format in csv, in the format of "name,lastModifiedInEpoch"
command = """find /var/log/tedge/agent -name "*.log" -exec stat -c '%n,%Y' {} \;"""

steps = [
    { script = "main.js", config = {olderThan = 86400} },     # filter by time, and strip the time from the message
]

[output]
process = """tedge upload c8y --file ${payload}"""

Describe alternatives you've considered

Solution 1: Allow a process to be executed as part of a step

name = "file-uploader"
version = "1.0.0"
description = "Upload all files within a folder"

[input.process]
# list log files and print their format in csv, in the format of "name,lastModifiedInEpoch"
command = """find /var/log/tedge/agent -name "*.log" -exec stat -c '%n,%Y' {} \;"""

steps = [
    { script = "main.js", config = { olderThan = 86400 } },     # filter by time, and strip the time from the message
    { process = "tedge upload c8y --file ${payload.message}" },
]

Additional context

[didier-wenzek]

I would really avoid to add effect-full processing steps, because this will break the core concept of flows being about message transformation.

Processing flows output using arbitrary commands makes more sense (and is definitely better than the contrived example where a second flow is used for its input acting as a command executor for a first flow). However, we will have to handle properly timeout and error management.