Operator functions

Operator contracts invoked by generated Python streams. Implement only the contract selected by the stream type in the designer.

When to use

Use this page while implementing a generated business function or endpoint handler.

Behavior

Every invocation receives the current stream and input value; runtimes with explicit message contexts also pass the context separately.
Emission is explicit. Returning from Map, FlatMap, KeyBy, or Process without using an output intentionally produces no downstream value.
Process has separate normal and error outputs. Throwing or returning a runtime error is not a substitute for intentionally emitting the graph error type.
Join inputs are already grouped by key and retained by the configured join storage before the user callback runs.

Streams without a user callback

Input, Sink, Merge, Split, FlatMap Iterable, Cycle Link, Error, and individual When branches are primarily graph/runtime nodes. Their behavior is generated from topology and configuration; they do not all require a custom operator implementation.

Output and error semantics

Normal output follows persisted graph links. A Process error output follows the dedicated Error link. A callback can emit more than once unless the selected stream contract or downstream protocol imposes a stricter rule.

Operator contracts

User-facing methods selected by generated stream wiring.

Type or methodRoleDescription

MapFunction[T, R].map(stream, value, out)Async Protocol

Transforms one input value and emits the resulting value through the normal output.

FilterFunction[T].filter(stream, value) -> boolAsync Protocol

Returns whether the current value should continue to downstream streams.

FlatMapFunction[T, R].flatmap(stream, value, out)Async Protocol

Emits zero, one, or many output values for one input value.

KeyByFunction[T, K, V].key_by(stream, value, out)Async Protocol

Creates a key/value pair used by keyed operators and join storage.

ProcessFunction[T, R, E].process(stream, value, out, err_out)Async Protocol

Runs general business logic with independent normal and error outputs.

JoinFunction[K, T1, T2, R].join(stream, key, left_values, right_values, out) -> boolAsync Protocol

Receives the current key and the buffered values from the left and right inputs, emits results, and returns the framework completion decision for that key.

MultiJoinFunction[K, T, R].multi_join(stream, key, values, out) -> boolAsync Protocol

Receives the current key and one buffered value collection per input, emits results, and returns the framework completion decision for that key.

DelayFunction[T].duration / delay_errorAsync Protocol

Calculates how long the value should wait; the paired error method decides what to emit when scheduling the delay fails.

BuildSwitchFunction[T].build_switch(stream, when_items)Case routing

Builds the runtime selector used by a Case stream to choose one of its typed When branches.

Generated extension pattern

async def process(self, stream, value, out, err_out) -> None:
    await out.out(value)