Collect

A fan-in point. Every wire arriving at this node gets a slot; once every slot is filled, all of them are forwarded together as a single message.

Inputs

payload any

That branch's contribution to the set — whatever the wire delivered. Another property can be collected instead, or the whole message; see Combine.

topic string

That branch's key when create is a key/value Object. Any other property can be named under keyed by. A branch carrying no such property is keyed by the id of the node it arrived from rather than being dropped.

Nothing else has to be set. A message from a HealthConnect node also carries ch, and errorInfo when that node failed; both are written by the node that sent them rather than by you, and both are read here automatically — see The outgoing message and Failing a set.

Outputs

payload any

The branches combined, in wire order rather than arrival order, so the result is stable. The shape follows Mode. Under combine the branches the result is written back to the property it was collected from, so naming a property other than payload in Combine puts the result there instead; a reduce collects from no one property, so its result always goes to payload.

ch.collect.results array

The same wires in the same order, each entry pairing the payload with source (the id of the node the wire left, or null if it could not be attributed) and ch — whatever that branch alone put in the namespace, present only when it put something there.

Details

A message fills the slot belonging to the wire it arrived on, replacing whatever that slot held before. Nothing is sent until every wire has delivered at least once.

That holds however many triggers feed the node. Where one trigger feeds three wires and another feeds a fourth, nothing is sent until both triggers have run.

If the same trigger runs again before the other one does, its wires simply overwrite their own slots — the newest value from each wire is the one sent. Once a set goes out the slots are emptied, ready for the next round.

The outgoing message

The outgoing ch holds collect and nothing else. A fan-in has no single incoming namespace to carry forward — two FHIR Read nodes meeting here both write ch.fhir.read — so each branch is kept separately under results rather than one of them overwriting the other. Reach for ch.collect.results[0].ch.fhir.read, not ch.fhir.read.

Properties outside branch-scoped ch and errorInfo are taken from the last message in, the same convention Node-RED's own join node uses. An error collected from one branch is not promoted into an error on the completed set.

Only ch.collect.source is this node's own bookkeeping, so collectors chain: feed one into another and the inner set arrives as that branch's ch, at ch.collect.results[n].ch.collect.results.

Combining the branches

Mode and the fields under it shape the combined property only. Whichever is chosen, ch.collect.results still holds every branch separately, so nothing is lost by the choice.

The two modes are Node-RED's own join node's, named and behaving as join names and behaves them, so what is learned about one node carries to the other. join's third mode, automatic, is not offered: it reads msg.parts to learn what a complete sequence is, and the messages arriving here are branches of a fan-out rather than parts of a split sequence. This node learns what a complete set is from its own wiring instead.

combine the branches

Combine names what one branch contributes — a message property, payload by default, or the complete message — and create names the shape they are built into.

an Array

One entry per inbound wire, in wire order. The default.

a key/value Object

{ key: value }, the key read from msg.topic by default — the same property join uses, and any other can be named under keyed by. Where two branches carry the same key the later wire wins. A branch carrying no such property takes the id of the node it arrived from as its key, warned about once; join drops such a message, which a fan-in cannot do without silently losing a wire.

a merged Object

The branch values' own keys copied into a single object, later wires winning. A branch whose value is not an object cannot be merged; it is left out with a warning, and is still in ch.collect.results.

a String or a Buffer

The branch values run together with joined using between them. A branch whose value cannot go into a buffer is left out with a warning rather than failing the set, again leaving ch.collect.results untouched.

reduce the branches

A JSONata expression run once per branch over an accumulator, exactly as join's reduce runs it. The bindings are join's, so an expression written for one node works in the other:

$A

The accumulator so far, starting at Initial value.

$I

The branch's own position in wire order, counting from 0. It stays that position when Evaluate in reverse order is ticked — reversing changes the order the branches are visited in, not what any of them is.

$N

The number of branches in the set.

The expression is evaluated against the branch's own message, so payload reads that branch's payload. Summing them is $A + payload from an Initial value of 0.

Fix-up exp, if given, runs once at the end over the finished accumulator, with $A and $N bound and no message to read — $A / $N turns that sum into a mean.

The node checks the expressions when the flow is deployed. One that cannot be read, or a missing Reduce exp, shows a red status and fails any message that reaches the node, rather than looking ready and then failing every set it completes. An expression that fails while running fails that set, naming the error.

Failing a set

A branch that reports its failure

A HealthConnect node that fails does not stop the flow — it sends a message on carrying errorInfo and a payload.success of false. That message reaches this node like any other and fills its wire's slot.

With On fail set to fail the whole set, the node waits for every wire as usual and then, if any of them failed, forwards nothing and raises an error naming each failed branch. The error carries the message the set would have been sent as, so a Catch node receives the full ch.collect.results and can see exactly which branch failed and why.

Set it to collect it like any other result and the failure is simply collected, for a flow that means to inspect the failures downstream.

The set is failed only once every wire has delivered, never on the first failure. Releasing the set early would leave the branches still in flight to land in an emptied node and open a phantom set.

A branch that fails silently

Not every failure arrives as a message. A node that fails by throwing — a core function node, say — never reaches its send, so its wire delivers nothing at all and its slot would stay empty for ever. The node sits at 3/4 and the set is never released.

The node catches this by itself. The runtime reports a failing node with its id, and that id is one this node already knows from its wiring, so the empty slot is filled with a payload of { success: false, error } and an errorInfo marked silent. The set then completes and On fail decides what happens to it, exactly as for a branch that reported its own failure — one setting governs both kinds of failure. A warning naming the branch is logged either way.

A branch that sends its failure and then fails is left alone. Every HealthConnect node does exactly that — the failure message goes on the wire first, then the node completes with the error — so both settings would otherwise fire for the same branch. The node knows a message is on its way before the sender has finished, so it waits for the real one rather than inventing a substitute. What you collect is the branch's own error, with its own text and its own ch contributions.

Only the node on the wire is visible this way. Where the throw happens further upstream — trigger → A → B → Collect and A throws — B never receives anything and so never completes, nothing is reported for it, and the set still waits. Use Timeout for that.

Giving up

Set Timeout to act when no further message has arrived for that long. The timer measures quiet, not total elapsed time: it restarts each time a slot fills. Left at 0 the node waits indefinitely, which is right when every wire is certain to fire eventually.

On timeout then decides what happens: fail the incomplete set forwards nothing and raises an error saying how many wires never delivered, or send on whatever arrived releases the partial set with a warning. Either way the slots are emptied and the next round starts clean.

A set that both timed out and holds a failed branch is reported as having timed out: an incomplete set is the more fundamental fault, and the wires still outstanding may be the ones that would have failed. A reduce expression that failed is reported last of the three, for the same reason — the set was never going to be sent whatever the expression did with it, and the branch that broke the expression may well be the one that failed.

When it will not run

The node reads its wiring when the flow is deployed. If it cannot find any inbound wires it shows a red No inbound wires status and fails any message that reaches it, rather than passing messages straight through and looking like a collector sending sets of one.

This is also why it cannot be used inside a subflow: a node running in a subflow instance is given a fresh id that does not match its own entry in the deployed flow, so its wires cannot be found. Put it in a normal flow.

Known limitations

Slots are per wire, not per run. Two runs overlapping will overwrite each other's slots rather than being kept apart — which is what makes the overwrite behaviour above possible, but means this node is not the right choice when several runs must stay independent.

A wire that delivers several messages for one run keeps only the last.

A message that reaches the node other than along one of its wires — injected into it from the editor, say — cannot be attributed to a slot. All such messages share a single slot, logged once as a warning, and go last in the output with a source of null. That slot counts towards the total, so a stray message completes the set in place of a wire that has not delivered yet, rather than being collected alongside it.

A reduce expression is evaluated asynchronously, because it may reach into flow or global context. The slots are emptied before it starts, so a message arriving while it runs begins the next set rather than joining the one being reduced — but if that next set completes first, it is sent first. Sets are combined in the order they complete, not necessarily the order they are released.

A node deployed before On fail and On timeout existed carries neither setting, and keeps the older behaviour — collecting failures, releasing partial sets — so that flows already running are not changed underneath them. A newly added node defaults to failing in both cases. Opening an older node and saving it takes up the newer defaults. A node deployed before Mode existed carries no mode and goes on combining the branches as it did.

Examples

Two parallel reads rejoined:

             ┌──────────────────┐
          ┌─▶│ FHIR Read        │─┐
┌────────┐│  └──────────────────┘ │  ┌────────────┐   ┌──────────┐
│ inject │┤                       ├─▶│ Collect    │──▶│ function │
└────────┘│  ┌──────────────────┐ │  └────────────┘   └──────────┘
          └─▶│ Stedi Eligibility│─┘
             └──────────────────┘

Nothing leaves Collect until BOTH wires have delivered.

What each branch sends:

// from the FHIR Read wire
msg.payload = { resourceType: "Patient", id: "synthetic-1" };
msg.ch      = { fhir: { read: { id: "synthetic-1" } } };

// from the Eligibility wire
msg.payload = { eligible: true };

Output — Mode combine the branches, create an Array:

{
    "payload": [
        { "resourceType": "Patient", "id": "synthetic-1" },
        { "eligible": true }
    ],
    "ch": {
        "collect": {
            "results": [
                {
                    "source": "fhir_read_1",
                    "payload": { "resourceType": "Patient", "id": "synthetic-1" },
                    "ch": { "fhir": { "read": { "id": "synthetic-1" } } }
                },
                {
                    "source": "eligibility_1",
                    "payload": { "eligible": true }
                }
            ]
        }
    }
}

Output — Mode reduce the branches:

Reduce exp     $merge([$A, payload])    Initial value  {}   (json)
→ msg.payload = { "resourceType": "Patient", "id": "synthetic-1", "eligible": true }

Reduce exp     $A + payload             Initial value  0    (num)
→ msg.payload = the branch payloads summed

Reduce exp     $A + payload             Fix-up exp     $A / $N
→ msg.payload = their mean

Best Practices

• Read a branch's namespace at ch.collect.results[n].ch.fhir.read, never ch.fhir.read — the outgoing ch holds collect and nothing else.
• Set a Timeout on any flow where a branch can hang. Left at 0 the node waits for ever, and a wire that never delivers holds the set for ever with it.
• For a plain merge use create → a merged Object rather than a reduce. It needs no expression and cannot be got wrong.
• In a reduce, always fold the accumulator in: $merge([$A, payload]), never $merge(payload) — the latter reads no accumulator, so only the last branch survives.
• Give every branch its own wire into this node. Two branches sharing one feeder share one slot, and the later one overwrites the earlier.
• Put it in a normal flow, never inside a subflow — it finds its wires by reading the deployed flow and a subflow instance's ids do not match.
• Leave On fail on fail the whole set unless the flow genuinely means to inspect failures downstream; a collected failure otherwise reads like data.