Sink Collect
Closes a source stream and collects its items into a single message.
A source stream turns one message into many, each carrying a single item. Sink Collect does the reverse: it waits for every item in the stream to arrive, gathers them into an array, and emits one message containing that array.
Use a Sink Collect node in your Flow wherever a stream needs to become a single result again — before returning an HTTP response, before writing a summary record, or before any node that expects one message rather than many.
See Choosing between Source From, Loop, and Split/Join on the Source From page for how streams relate to other iteration approaches in a Flow.
Sink Collect is the closing half of a pair. It is used with any node that opens a source stream — a Source From node, or a node running in Source mode such as a Query Object node with Unfold all pages to source enabled. A node that opens a stream without reaching a Sink Collect is reported as a lint error in the editor.
To configure a Sink Collect node in the Flow Editor, drag it from the Stream group in the Palette to the Workspace canvas, then double-click the node to open and configure its behavior.
Properties
Name — optional display name for the node in the editor. By default the node shows a label of "Sink Collect"; the value you provide here replaces it, making your flow easier to read.
Input — the property on each incoming stream message that holds the item to collect. This is a typed input and defaults to msg.payload.
Output — where the collected array is written on the outgoing message. This is a typed input and defaults to msg.payload.
Because Input and Output are set independently, a stream can be read from one property and the result written to another, leaving the original property untouched.
Output shape
Sink Collect always emits an array, in the order the items arrived. This holds even when the stream carries a single item. A one-item stream produces a one-element array, not the bare value.
Downstream nodes can therefore treat the output uniformly, without checking whether the stream carried one item or many.
The node emits exactly one message per stream. Nodes placed after it run once, not once per item.
Example
Continuing the /report endpoint from the Source From page, a weather application returns all of its stored readings in a single response:
Ten records enter the stream through Source From, each travelling as its own message with one reading at msg.payload. Sink Collect gathers them and writes the array back to msg.payload:
A Log node placed before Sink Collect shows ten separate entries. The same Log node placed after it shows a single entry containing an array of ten. That difference is the clearest way to confirm the stream opened and closed correctly.
The HTTP Response node at the end of the chain receives one message, so the caller gets one JSON array. Without Sink Collect there would be ten messages arriving at a node that can only respond once.
Collecting a stream from any source
Sink Collect does not need to be paired with a Source From node specifically. It closes a stream regardless of which node opened it.
Replacing Query Objects and Source From with a single Query Objects node in Source mode produces the same result:
With Unfold all pages to source enabled, the Query Objects node emits its records as a stream directly — its editor label changes to [Source] Query Objects — and Sink Collect gathers them into the same ten-element array. A Log node placed before Sink Collect shows ten entries; one placed after shows a single entry containing all ten.
Reshaping items before collecting
Any node placed between the stream source and Sink Collect runs once per item, so per-item work belongs there rather than after the collection point.
In the example above, each record carries a _metaData property that is larger than the weather data itself. Reducing each record to the fields the caller needs — before Sink Collect, while the items are still separate — keeps the collected array small, and is simpler than transforming an array of ten afterwards.
Last updated
Was this helpful?