Add proper streaming consumption support and document output of AsyncAPI subscribe operations

[matthias-pichler]

What would you like to be added?

Right now the DSL reference shows this example:

document:
  dsl: '1.0.0-alpha5'
  namespace: test
  name: asyncapi-example
  version: '0.1.0'
do:
  - subscribeToChatInboxForAmount:
      call: asyncapi
      with:
        document:
          endpoint: https://fake.com/docs/asyncapi.json
        operation: chat-inbox
        protocol: http
        subscription:
          filter: ${ .roomId == $workflow.input.chat.roomId } 
          consume:
            amount: 5
            for:
              seconds: 10

Assuming 3 messages are sent to this chat in the following 10 seconds ... what would be the output of this task?

Proposal(s):

I propose that the output of the task should be an array of messages received.

[{"roomId": "1", "text": "Hi"}, {"roomId": "1", "text": "Hello"}, {"roomId": "1", "text": "Bye"}]

But this restricts us to only have "buffered" consumption. It would be beneficial in my opinion to have proper streaming consumption.

Maybe something like a foreach (name is WIP) property that specifies a list of tasks to perform per message?

document:
  dsl: '1.0.0-alpha5'
  namespace: test
  name: asyncapi-example
  version: '0.1.0'
do:
  - subscribeToChatInboxForAmount:
      call: asyncapi
      with:
        document:
          endpoint: https://fake.com/docs/asyncapi.json
        operation: chat-inbox
        protocol: http
        subscription:
          filter: ${ .roomId == $workflow.input.chat.roomId } 
          consume:
            until: ${ true } # forever 😈
          foreach:
            - emitEvent:
                emit: {}

Alternative(s):

No response

Additional info:

No response

Community Notes

• Please vote by adding a 👍 reaction to the feature to help us prioritize.
• If you are interested to work on this feature, please leave a comment.

[cdavernas]

@matthias-pichler I love the proposal! BTW, the foreach could be extended to listen task, too. This, btw, would reintroduce an 0.8 feature which got lost in translation 😉