SubStream

When to use

Use a named reusable business stage inside one service, without adding a transport endpoint or exposing every implementation helper as a graph node.

Behavior

• Outgoing links send the input into ordinary operators. The existing source field binds the result-producing stream.
• The generated service exposes a named typed accessor. A custom maker can inject the service through an interface that declares only the accessors its business code needs.
• SubStreamCollector and SubStreamCollectorFunc provide the collector interface and function adapter. True completes the invocation; false continues collecting. There is no inferred result count or automatic completion when a branch returns. Cancellation or a deadline can end the wait.
• Each call binds its collector through a derived context. No message ID argument, graph copy, extra ResultStream node or SubStream error port is required.
• Callbacks within one invocation are serialized. Late results are discarded once that call completes.
• Completion ends result delivery, not all graph work. Existing Join keys, storage, pools and transport behavior remain unchanged.
• Business failures remain ordinary result values or graph error paths; they do not automatically become the error returned by Consume.

Create and wire a SubStream

Add a SubStream in the Stream page and select its owning service. Set Value Type to the input argument type T, then connect one ordinary body operator to the entry as its source.

Set Result Stream on the entry to the reachable body node that produces R. In YAML this property is named source; it is a return path, not a second invocation. The entry has no endpoint or business function.

A short body can be SubStream -> Map -> return to SubStream. More complex bodies use existing operators. Keep simple operations inside the business function instead of translating its conditions and loops into graph nodes.

Call from business code

Every declared SubStream is generated, even when calls from business functions are not drawn as graph edges. Inject only the typed accessors the business maker needs. Capture handles during construction, but call them after graph binding and while the service is alive.

For an entry named Lookup: Go uses Lookup().Consume(ctx, value, collector); TypeScript uses await getLookupSubStream().consume(context, value, collector); Python uses await get_lookup_substream().consume(value, collector). Rust uses get_lookup_substream(); C++ uses getLookupSubStream(). Both accept their runtime context, payload and collector types.

Python uses an async value-only collector and ContextVars rather than an explicit context argument. TypeScript accepts boolean or Promise<boolean>. Rust returns a runtime Result from its async call; C++ follows runtime exception conventions. Go returns an error. Business failures are not automatically converted into these runtime failures.

Completion, concurrent calls and nested calls

Return true after the expected count, business outcome or explicit end marker. Returning false means keep waiting; a Filter or branch that emits no result needs cancellation or a deadline to avoid an indefinite wait.

Concurrent and nested calls keep separate collectors without copying the graph. A collector receives its original caller context, including an enclosing SubStream binding. Callback execution is serialized per invocation.

Completion discards late results rather than forcibly stopping all graph work. An already-running callback must cooperate with cancellation so cleanup can finish. Shared Join keys and worker capacity are still your responsibility; isolated collectors do not isolate shared business state.

Temporal workflows

Go, Python and TypeScript use workflow-aware cooperative execution. Use the generated workflow environment, deterministic code and durable timers. External effects belong in Activities; do not introduce native blocking waits or ordinary wall-clock timers into workflow code.

Invocation state is reconstructed on replay. C++ and Rust do not support Temporal.

General properties

Identity, ownership and operator-specific mode.

Connections properties

Upstream graph references accepted by this operator.

Data Types properties

Message contracts visible at this node.

Execution properties

Pipeline placement and runtime policy.

Position properties

Designer-only canvas coordinates.

Configuration example

type: SubStream
name: SearchOffers
valueType: SearchRequest
source: combinedOffers
pipeline: offerSearch

Related documentation