transforms

Transforms run in the order they are listed, each taking a batch and returning zero or more batches — which is how splitter and reduce change a stream's cardinality. Everything is untyped JSON, and a transform addresses fields by field path, so _meta.subject is reachable wherever value is.

Two of these are a pair rather than two components: remember and recall share a state bucket, and they are separate because chain order is the semantics — remember is a tap that passes its batch on unchanged, and recall writes what was remembered onto the messages that come after.

Anything contradictory is refused when the pipeline is built rather than producing a strange message once per batch forever: a reducer with no aggregations, an as that would overwrite a group field, a map writing a path that runs through a scalar.

buffer

Holds messages back and hands them on when a trigger says to.

There are three triggers and they compose: a message count, a length of time, and a condition on a state bucket. Any of them is enough on its own — whichever comes first ends the wait, the same rule the input-level batch buffer follows. A buffer with no trigger at all fails to build.

size is the one that has always been here and it behaves exactly as it did: messages are handed on in batches of exactly that many, as they fill. The other two release everything currently held as a single batch, however much that is — which is the useful reading of "the run is finished, send what you have".

Distinct from the buffer option on an input: that one batches what an input produces, before any transform has seen it. This one sits in the chain, so it batches what the transforms in front of it produced — after a filter has thinned the stream, or a recall has enriched it.

field	type		description
max_messages	integer	optional	never hold more than this many messages: reaching it releases them all, whatever the triggers say, and says so in the log once. Required unless size is set, because size is its own bound — a buffer waiting on a condition that never comes true is otherwise a memory leak that grows at the rate of the stream.
seconds	integer	optional	release everything held this many seconds after the first held message. The window opens when a message is held rather than when the last batch went out, so this is a bound on how long a message waits and not a cadence — an idle buffer holds nothing and no clock is running.
size	integer	optional	hand messages on in batches of exactly this many, as they fill. On its own this is a buffer that only ever counts, and is what this transform has always done.
until	object	optional	release everything held when a state bucket says so. This is the trigger a different pipeline can pull: buckets are global, so one pipeline can mark a run complete and this one hands on what it gathered while the run was going.

until

field	type		description
conditions	list of numeric | string	required	what has to be true of that key for the buffer to be released. All of them, and at least one — a gate with no conditions would be a buffer that releases on every write to the bucket.
bucket	string	optional	which bucket to watch. Defaults to the one this pipeline's state names; a pipeline with no state of its own has to name it here.
key	string	optional	which key in that bucket to read. A literal key, not a field path — this is one gate for the whole buffer, so there is no message to take a key from. Leave it out for the bucket-wide value, which is what remember writes when its pipeline's state has no key.

until.conditions — each entry — type: "numeric"

field	type		description
field	string	required	the field to test — a dotted path, like anywhere else
operator	greater_than | less_than | equal_to	required	How a number is compared to the one in the config.
value	number	required

until.conditions — each entry — type: "string"

field	type		description
field	string	required	the field to test — a dotted path, like anywhere else
operator	equal_to | contains	required	How a string is compared to the one in the config.
value	string	required

http

Posts the batch to an http endpoint as a JSON array and replaces it with the JSON array in the response — so the service on the other end is the transform.

field	type		description
url	string	required	endpoint to send the batch to
verb	GET | POST | PUT | PATCH | DELETE	required	http method. Accepted but not honoured yet: every request is a POST.

splitter

Cuts one batch into several smaller ones — the opposite of buffer.

Note the current limitation: messages left over after the last whole chunk are dropped, so 4 messages with out_size: 3 emit one batch, not two.

field	type		description
out_size	integer	required	how many messages go in each emitted batch

reducer

Reduces a batch to one message per group, carrying whatever was asked for about it. Pair it with a buffer, or it will only ever see one message at a time.

With no group_by the whole batch is one group and one message comes out; with one, a message comes out per distinct combination of those fields, in the order the groups were first seen. The emitted message carries the grouping fields under their own names alongside the aggregations.

Each aggregation is a function, the field to apply it to and the as name the answer is written under — {"function": "avg", "field": "value", "as": "mean"}. count is the one function that needs no field: without one it counts the messages in the group, with one it counts the messages that carried it.

field	type		description
aggregations	list of aggregation	required	what to compute. At least one, and each needs a distinct as.
group_by	list of string	optional	the fields whose combination defines a group. Omit it to reduce the whole batch at once.
on_missing	error | skip	optional	what to do about a message missing one of the fields above

aggregations — each entry

field	type		description
function	sum | avg | min | max | count | count_distinct | first | last | collect | median | stddev	required	how to combine the values
as	string	required	the field the emitted message carries this answer under. Two aggregations may not share one, and none may collide with a group_by field.
field	string	optional	the field to aggregate. Required by every function except count, which counts messages when it is left out.

filter

Drops messages that don't match a condition, and drops the whole batch if none of them do. Pick either the Numeric or the String form — the fields differ because the comparisons do.

{"type": "filter", "Numeric": { … }}

field	type		description
field	string	required	the field to filter on
operator	greater_than | less_than | equal_to	required	How a number is compared to the one in the config.
value	number	required

{"type": "filter", "String": { … }}

field	type		description
field	string	required
operator	equal_to | contains	required	How a string is compared to the one in the config.
value	string	required

remember

Writes values from matching messages into the pipeline's state bucket, keyed by whatever the pipeline's state.key names.

The message itself is passed on unchanged — this is a tap on the stream, not a filter. A transform called remember that quietly swallowed what it remembered would be a surprise, and the message is usually still wanted.

Needs a state on the pipeline; it fails to build without one.

field	type		description
remember	list of object	required	what to take from a matching message. At least one, each with a distinct as.
when	list of numeric | string	optional	which messages to remember from — all of these have to match. Leave it out to remember from every message, which is right for a stream carrying one kind of thing and wrong for one carrying several.

remember — each entry

field	type		description
field	string	required	the field to take the value from
as	string	required	the name to remember it under, which is the name recall asks for it by. Two entries may not share one.

when — each entry — type: "numeric"

field	type		description
field	string	required	the field to test — a dotted path, like anywhere else
operator	greater_than | less_than | equal_to	required	How a number is compared to the one in the config.
value	number	required

when — each entry — type: "string"

field	type		description
field	string	required	the field to test — a dotted path, like anywhere else
operator	equal_to | contains	required	How a string is compared to the one in the config.
value	string	required

recall

Writes values from the pipeline's state bucket onto every message, under the names they were remembered by.

This is how a slow-moving fact — the unit being produced, the recipe in force — reaches the fast stream that has to be attributed to it. The values land as top-level fields, so a reducer downstream can group by them without knowing where they came from.

Needs a state on the pipeline; it fails to build without one.

field	type		description
recall	list of string	required	the names to read out of the bucket, as remember wrote them. Each one is written onto the message under the same name.
on_missing	skip | null | error	optional	what to do about a message whose key has nothing remembered under it yet

map

Rewrites the shape of every message: renames, promotions, constants, casts and projections, applied in order.

Each entry in mappings reads fields from the message and writes one field back, and later entries see what earlier ones wrote — so an intermediate value is just a mapping whose target a later mapping reads (and, under keep: all, a drop takes away again).

Reads are dotted paths, like everywhere else. Writes are too: an as of sensor.id puts the value inside a sensor object, creating it if it isn't there.

The message is passed through unchanged, with the mappings laid over it, unless keep says otherwise. One message always comes out — this never drops one, and never makes two. Reach for filter or splitter for those.

field	type		description
mappings	list of copy | constant | coalesce | cast | concat | arithmetic | drop	required	what to write, in the order it is written. At least one, and no two may write the same field.
keep	all | mapped	optional	whether fields nothing mapped survive
on_missing	error | omit | null	optional	what to do about a message missing a field a mapping reads. A default on the mapping itself is answered first, and is the better way to say that one particular field is expected to be absent.

mappings — each entry — type: "copy"

field	type		description
from	string	required	the field to read — a dotted path, like anywhere else
as	string	optional	the field to write. Left out, it is from's last segment, which is the reading that makes promoting a nested value the short spelling.
default	text | number | boolean | null	optional	what to write when from isn't there, instead of applying on_missing

mappings — each entry.default — type: "text"

field	type		description
value	string	required	the text

mappings — each entry.default — type: "number"

field	type		description
value	number	required	the number

mappings — each entry.default — type: "boolean"

field	type		description
value	boolean	required	the flag

mappings — each entry.default — type: "null"

This component takes no configuration.

mappings — each entry — type: "constant"

field	type		description
value	text | number | boolean | null	required	the value to write
as	string	required	the field to write it to

mappings — each entry.value — type: "text"

field	type		description
value	string	required	the text

mappings — each entry.value — type: "number"

field	type		description
value	number	required	the number

mappings — each entry.value — type: "boolean"

field	type		description
value	boolean	required	the flag

mappings — each entry.value — type: "null"

This component takes no configuration.

mappings — each entry — type: "coalesce"

field	type		description
from	list of string	required	the fields to try, in order. At least two — with one, this is a copy.
as	string	required	the field to write the first value found to
default	text | number | boolean | null	optional	what to write when none of them is there

mappings — each entry.default — type: "text"

field	type		description
value	string	required	the text

mappings — each entry.default — type: "number"

field	type		description
value	number	required	the number

mappings — each entry.default — type: "boolean"

field	type		description
value	boolean	required	the flag

mappings — each entry.default — type: "null"

This component takes no configuration.

mappings — each entry — type: "cast"

field	type		description
from	string	required	the field to read
to	text | integer | float | boolean | timestamp | date | uuid | json	required	what to convert it to
as	string	optional	the field to write. Left out, it is from's last segment — so casting a field in place is {"from": "value", "to": "float"}.
default	text | number | boolean | null	optional	what to write when from isn't there. A value that is there and won't convert is an error either way — that is a stream that isn't what the config says it is, not a missing field.

mappings — each entry.default — type: "text"

field	type		description
value	string	required	the text

mappings — each entry.default — type: "number"

field	type		description
value	number	required	the number

mappings — each entry.default — type: "boolean"

field	type		description
value	boolean	required	the flag

mappings — each entry.default — type: "null"

This component takes no configuration.

mappings — each entry — type: "concat"

field	type		description
parts	list of field | value	required	the pieces, in order. At least one.
as	string	required	the field to write the joined string to

mappings — each entry.parts — each entry — type: "field"

field	type		description
field	string	required	the field to read

mappings — each entry.parts — each entry — type: "value"

field	type		description
value	string	required	the text

mappings — each entry — type: "arithmetic"

field	type		description
left	field | value	required	the left-hand operand
operator	add | subtract | multiply | divide	required	what to do with them
right	field | value	required	the right-hand operand
as	string	required	the field to write the answer to

mappings — each entry.left — type: "field"

field	type		description
field	string	required	the field to read — it has to hold a number

mappings — each entry.left — type: "value"

field	type		description
value	number	required	the number

mappings — each entry.right — type: "field"

field	type		description
field	string	required	the field to read — it has to hold a number

mappings — each entry.right — type: "value"

field	type		description
value	number	required	the number

mappings — each entry — type: "drop"

field	type		description
from	list of string	required	the fields to remove. At least one.

script

Runs a rhai script over each message, or over the batch as a whole, and emits whatever the script asks for.

A script reaches the message as msg, and emits with emit(value) — zero times to drop it, once to replace it, many times to split it. That covers filter, map and splitter in one, which is the point: what a script is for is the case none of those three reach.

The script is compiled when the pipeline is built, so a syntax error is a pipeline that refuses to start rather than one that fails every batch forever — the same rule the reducer's build-time checks follow. What cannot be checked until a message arrives (a field that isn't there, a type that won't convert) fails that batch and shows up on the card.

Every script runs under an operation budget. That is not a tuning knob with a safe default, it is what makes this component safe to have: the script runs synchronously inside the run loop's task, so a script that loops forever would wedge a worker thread rather than merely breaking its own pipeline.

A script may import other rhai files — shared helpers, written once — by a literal path relative to the config file's directory, which it may not climb out of; the .rhai extension is implied. Imports resolve when the pipeline is built, so a broken one refuses to start rather than failing batches, and a running script never touches the filesystem.

field	type		description
source	inline | file	required	the script itself, written inline or kept in a file beside the config
max_operations	integer	optional	how many rhai operations one run of the script may take before it is stopped and the batch failed. Leave it out for the default, which is generous for anything that isn't looping by mistake; raise it for a script that legitimately walks a large array.
scope	message | batch	optional	whether the script sees one message at a time or the whole batch

source — type: "inline"

field	type		description
code	rhai script	required	the rhai source

source — type: "file"

field	type		description
path	string	required	the path, relative to the config file's directory. It may not climb out of that directory.