C++ Coro Operator functions

Operator contracts invoked by generated C++ Coro 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

operator()(MessageContext, StreamBase&, const T&, Output&&) -> awaitable<void>Map / FlatMap / KeyBy

Await out.out(context, value) for every emitted result; do not detach a new coroutine at each graph edge.

operator()(MessageContext, StreamBase&, const T&) -> boolFilter predicate

A synchronous predicate can return its decision directly. Follow the signature generated for the operator.

operator()(MessageContext, StreamBase&, const T&, Output&&, ErrorOutput&&) -> awaitable<void>Process

Await the selected normal or error output so completion and backpressure remain part of the call.

operator()(context, stream, key, values, out) -> awaitable<bool>Join / MultiJoin

Await output emission; co_return true to complete the key, or false to retain accumulated values.

operator()(context, stream, value) -> durationDelay calculation

Calculating a delay does not need to suspend. Scheduling and later delivery belong to the runtime.

Case / WhenConditional routing

Use the generated selector contract; each index selects an existing When branch.

Generated extension pattern

template <typename Output>
boost::asio::awaitable<void> operator()(
    servicelib::MessageContext context,
    servicelib::StreamBase&, const Order& value, Output&& out) const {
  co_await out.out(std::move(context), value);
}