| title | SubFlow |
|---|---|
| slug | /concepts/primitives/subflow |
A SubFlow is a Flow run that a parent Flow starts from one of its Steps. The child is a normal Flow with its own Steps, Attributes, and Channels. The parent waits for it like any other condition in WaitFor.
SubFlow has two main uses:
- Move complicated logic into a smaller Flow. This makes business logic easier to maintain.
- Fan out work. One Flow run cannot do too many operations at once, but it can hand work to many SubFlows. The SubFlows run at the same time, so the parent reaches much higher concurrency.
Dex Server computes the SubFlow ID from the parent Flow ID, the Step execution, and the condition index. So a retried or repeated start finds the same child instead of starting a second one. By default, the parent keeps waiting for a running child, uses the result of a completed child, and restarts a child that ended abnormally.
The SubFlow ID joins its parts with a dash. So a Step whose Step type contains a dash cannot start SubFlows. Otherwise two parents could compute the same SubFlow ID and share one child. Dex fails such a WaitFor attempt without retry. Give every Step that starts SubFlows a Step type without a dash.
When you start the parent Flow ID again, the new run usually computes the same SubFlow IDs as the earlier run. Dex then applies the reuse policy. With the default policy, the new run gets the earlier run's child, running or completed, until Dex deletes that child. To start a fresh child in every run, use ALWAYS_RESTART. It terminates the earlier child if that child is still running. When both children must keep running, use a different parent Flow ID.
A child keeps running when its parent Step is canceled or its parent Flow closes. Stop it yourself, or set a Flow timeout on the child so it ends on its own.
The target Flow must be registered on the Worker. Its start Step must accept the input passed to SubFlow. WaitFor starts the child. Execute reads its result after the child completes.
<SdkTabs examples={{ python: 'examples/python/dex_examples/primitives/subflow/parent_flow.py', go: 'examples/go/primitives/subflow/parent_flow.go', java: 'examples/java/src/main/java/io/superdurable/dex/primitives/subflow/SubFlowParentFlow.java', typescript: 'examples/typescript/src/primitives/subflow/subflow-flow.ts', rust: 'examples/rust/src/primitives/subflow/flow.rs', }}>
class SubFlowParent(Step[int]):
def __init__(self, target: Flow[int]) -> None:
self.target = target
self.options = SubFlowOptions(
timeout=timedelta(hours=1),
timeout_policy=FlowTimeoutPolicy.CANCEL,
)
def wait_for(self, context: Context, input: int) -> Wait:
return Wait.until(SubFlow.run(self.target, input, self.options))
def execute(self, context: Context, input: int) -> StepDecision:
result = SubFlow.get_condition_results(context)
output = result.single_output(int)
return graceful_complete(f"{SubFlow.get_flow_id(context)}|{output}")type subFlowParent struct {
dex.StepDefaults
child *SubFlowChildFlow
}
func (step subFlowParent) WaitFor(_ dex.Context, input int) (*dex.Wait, error) {
timeout := time.Hour
return dex.Until(dex.SubFlow(step.child, input, dex.SubFlowOptions{
Timeout: &timeout,
TimeoutPolicy: dex.TimeoutCancel,
})), nil
}
func (subFlowParent) Execute(ctx dex.Context, _ int) (*dex.StepDecision, error) {
result, err := dex.SubFlowResult(ctx)
if err != nil {
return nil, err
}
var output int
if err := result.DecodeSingleOutput(&output); err != nil {
return nil, err
}
flowID, err := dex.SubFlowID(ctx)
if err != nil {
return nil, err
}
return dex.GracefulComplete(fmt.Sprintf("%s|%d", flowID, output)), nil
}static final class SubFlowParent implements Step<Integer> {
private final SubFlowOptions options = SubFlowOptions.newBuilder()
.timeout(Duration.ofHours(1))
.timeoutPolicy(FlowTimeoutPolicy.CANCEL)
.build();
@Override
public Class<Integer> getInputType() {
return Integer.class;
}
@Override
public Wait waitFor(final Context context, final Integer input) {
return Wait.until(SubFlow.run(SubFlowChildFlow.class, input, options));
}
@Override
public StepDecision execute(final Context context, final Integer input) {
final Integer output =
SubFlow.getConditionResults(context).getSingleOutput(Integer.class);
return StepDecision.gracefulComplete(SubFlow.getFlowId(context) + "|" + output);
}
}class SubFlowParent implements Step<number> {
public readonly inputCodec = doubleCodec;
public constructor(private readonly target: SubFlowChildFlow) {}
public getStepType(): string {
return "SubFlowParent";
}
public waitFor(_context: Context, input: number): Wait {
return Wait.until(
SubFlow.run(this.target, input, {
timeoutMs: HOUR_MS,
timeoutPolicy: FlowTimeoutPolicy.CANCEL,
}),
);
}
public execute(context: Context, _input: number): StepDecision {
const result = SubFlow.getConditionResults(context);
const output = result.singleOutput(doubleCodec);
const flowId = SubFlow.getFlowId(context);
return gracefulComplete(`${flowId}|${output}`);
}
}#[derive(Default)]
struct SubFlowParent;
impl Step for SubFlowParent {
type Input = i32;
fn wait_for(&self, _context: &mut Context, input: Self::Input) -> HandlerResult<Wait> {
let child = SubFlowChildFlow::default();
let condition = SubFlow::run_with_options(
&child,
input,
SubFlowOptions::new()
.timeout(Duration::from_secs(3600))
.timeout_policy(FlowTimeoutPolicy::Cancel),
)
.map_err(HandlerError::from_error)?;
Ok(Wait::until(condition))
}
fn execute(&self, context: &mut Context, _input: Self::Input) -> HandlerResult<StepDecision> {
let result = SubFlow::condition_result(context)?;
let output: i32 = result.single_output().map_err(HandlerError::from_error)?;
let flow_id = SubFlow::flow_id_at(context, 0)?;
Ok(StepDecision::graceful_complete(format!(
"{flow_id}|{output}"
)))
}
}Use allOf to make the parent wait for all SubFlows to complete.
Use anyOf when the parent does not need to wait for all of them. A SubFlow keeps running after WaitFor stops waiting. Execute can still get the SubFlow ID to stop it, or pass the ID to another Step to decide how to handle it. See SubFlow results for combining SubFlows with other conditions.
The SubFlow page of each SDK reference lists the API names in that language.
SubFlowOptions accept the same options as starting a Flow from a Client, plus a reuse policy. See SubFlowOptions and SubFlowReusePolicy for the options and the reuse policy for each state of an existing child.