Aggregate Latest

Use this node to aggregate the latest attribute or time‑series values of an entity’s related entities and emit the result as an outgoing telemetry message for that entity — for example, computing the average temperature across all sensors located in a building and posting it to the building asset whenever a sensor reports new data, or counting the active devices related to a fleet asset. The node is event‑driven: every incoming message triggers a fresh aggregation pass for the message originator. To protect the database from bursts, the node deduplicates calculations per originator within a configurable window.

Source entity (originator). The originator of the incoming message. The aggregation result is emitted as outgoing telemetry for this entity.

Related entity. An entity reached from the source via a single‑hop relation matching the configured Direction and Relation type. The aggregator iterates over these related entities and reads their attributes / latest time‑series values.

Mapping. One (source key on related entity → aggregation function → target key on the originator) rule. A node can carry several mappings; each runs independently per incoming message.

Deduplication period. A per‑originator cool‑down window. If new messages for the same originator arrive while the window is open, only the latest one is kept; when the window closes, exactly one calculation is run for that originator. Prevents one‑message‑in / one‑calc‑out blow‑ups when many sensors update at once.

The fields below follow the order of the rule node editor.

A list of mapping rows. Add rows with + Add, edit with the pencil icon, remove with the cross icon. Each row defines one (source → aggregation → target) rule applied to every related entity for every incoming message.

Optional. Selects a rule engine queue for outgoing messages. If empty, the Main queue is used.

Optional free‑text note shown in the rule chain editor. Has no effect on processing.

For each incoming message:

1. Look up the in‑flight state for originator = msg.getOriginator().
   • If no calculation has been run for this originator yet, or the previous one happened more than Deduplication period seconds ago, start a calculation immediately with this message.
   • Otherwise, stash the message as the pending message for this originator and schedule a calculation at the next deduplication boundary. Any further messages for the same originator that arrive during the window simply replace the pending message — only the latest one is processed.
   • The incoming message is acked in either case.
2. When a calculation runs, the node:
   • resolves the related entities by querying relations of the configured Relation type in the configured Direction starting from the originator;
   • prefetches all required attributes / latest time‑series in one batch (per scope and per latest‑ts key list collected from all mappings);
   • for each mapping row, drops related entities for which the optional Entity filter returns false, applies the aggregation function across the rest using the configured Default value for missing sources, and writes the result under Target telemetry in the outgoing payload;
   • emits one outgoing message of the configured Output message type, originated by the source entity, on the Success branch.
3. Inactive originator entries (no message for ≥ 3 × Deduplication period) are evicted from the in‑memory map by an internal cleanup tick.

The node itself does not write telemetry — outgoing messages must be routed to a Save Time Series node downstream. Each calculation produces one outgoing message per originator, regardless of how many incoming messages were merged inside the deduplication window. Lowering the deduplication period multiplies write volume; raising it caps the calc rate at one per period per originator.

Per calculation the node also reads from the database: relations for the originator, plus latest time‑series and attribute lookups for every related entity (batched per scope across all mappings). When you have many mappings on the same scope, the cost is paid once for the prefetch and reused across mappings.

Configuration: Direction = From, Relation type = Contains, Deduplication period = 10 sec, one mapping with Latest telemetry checked, Source = temperature, Aggregation function = Average, Target = latestAvgTemperature. Place the node downstream of the rule chain branch that processes building telemetry.

Trigger: the building asset receives any message (typically a telemetry update from one of its sensors, routed via Originator Attributes / relations propagation).

Result: at most once every 10 seconds per building, one outgoing message is emitted, originated by the building, with:

{
  "latestAvgTemperature": 22.3
}

Wire Success into a Save Time Series node to persist the rolling building‑level average alongside per‑sensor readings.