等待一半完成
启动 N 个 SubFlow，等待其中一半完成，然后停止尚未完成的子 Flow。
每个子 Flow 都在独立的 SubFlowStep execution 中运行。子 Flow 完成后向 SubFlowCompletedCh 发布一条消息。WaitSubFlows 等待其中一半消息，然后为每个可能尚未完成的子 Flow 发布一条 AllDoneCh 消息。
关键点是父 Flow 达到阈值后不会立即结束。每个尚未完成的 SubFlowStep 会先消费一条 AllDoneCh 消息，取得自己的 SubFlowID 并停止对应子 Flow，然后才 graceful complete。这样每个未完成分支都有机会清理自己的 SubFlow。
核心实现
class WaitForHalfInit(Step[list[str]]):
def execute(self, context: Context, requests: list[str]) -> StepDecision:
if not requests:
return graceful_complete()
return go_to_many(
StepMovement.of(WaitSubFlows, len(requests)),
*(StepMovement.of(SubFlowStep, request) for request in requests),
)
class SubFlowStep(Step[str]):
def __init__(
self,
client_provider: Callable[[], AsyncClient],
example_subflow: ExampleSubFlow,
subflow_completed_ch: Channel[bool],
all_done_ch: Channel[bool],
) -> None:
self.client_provider = client_provider
self.example_subflow = example_subflow
self.subflow_completed_ch = subflow_completed_ch
self.all_done_ch = all_done_ch
def wait_for(self, context: Context, request: str) -> Wait:
return Wait.any_of(SubFlow.run(self.example_subflow, request), self.all_done_ch.for_one())
async def execute(self, context: AsyncContext, request: str) -> StepDecision:
if SubFlow.get_condition_results(context).status is not FlowStatus.RUNNING:
self.subflow_completed_ch.publish(context, True)
return graceful_complete()
await self.client_provider().stop_flow(SubFlow.get_flow_id(context))
return graceful_complete()
class WaitSubFlows(Step[int]):
def __init__(
self, subflow_completed_ch: Channel[bool], all_done_ch: Channel[bool]
) -> None:
self.subflow_completed_ch = subflow_completed_ch
self.all_done_ch = all_done_ch
def wait_for(self, context: Context, total: int) -> Wait:
return Wait.until(self.subflow_completed_ch.for_n((total + 1) // 2))
def execute(self, context: Context, total: int) -> StepDecision:
for _ in range(total - (total + 1) // 2):
self.all_done_ch.publish(context, True)
return graceful_complete()
class WaitForHalfParentFlow(Flow[list[str]]):
subflow_completed_ch = Channel("SubFlowCompletedCh", bool)
all_done_ch = Channel("AllDoneCh", bool)
def __init__(
self,
client_provider: Callable[[], AsyncClient],
example_subflow: ExampleSubFlow,
) -> None:
self.init = WaitForHalfInit()
self.subflow = SubFlowStep(client_provider, example_subflow, self.subflow_completed_ch, self.all_done_ch)
self.wait_subflows = WaitSubFlows(self.subflow_completed_ch, self.all_done_ch)
def get_steps(self) -> StepList[list[str]]:
return StepList.start_step(self.init).other_steps(self.subflow, self.wait_subflows)
def get_persistence_schema(self) -> PersistenceSchema:
return PersistenceSchema.of(self.subflow_completed_ch, self.all_done_ch)
例子: examples/python/dex_examples/patterns/parallel-subflows/wait_for_half_parent_flow.py