等待

本页介绍比基础功能更高级的 WaitFor 功能。

Until、AnyOf 和 AllOf 不要求 condition ID。通过 Channel 定义读取 Channel 结果，并通过 Context 检查 Timer 结果。仅当其他 API 有要求时才添加 ID，例如 AnyCombinationOf，或按 ID 选择 Timer。

ConditionResults

WaitFor 完成后，Execute 可读取满足 Wait 的内容。对 Channel，用 GetConditionResults / results / condition_results 读取已发布消息。对 SubFlow，用 SubFlowResult / getConditionResults / condition_result 解码子 Flow 输出。

AnyOf 的各备选条件之间不会 greedy 地消费 Channel。Dex 只消费被选中用来满足 Wait 的 Channel condition。即使其他 Channel condition 也已满足，它们也不会消费消息，消息会继续留在队列中。

当多个 AnyOf 备选条件在同一次求值中均已满足时，Dex 按规范顺序选择第一个可行候选： 先按声明顺序检查 Timer condition，再按声明顺序检查 Channel condition，最后按声明顺序检查 SubFlow condition。前面尚未满足的 condition 不会阻塞后面已满足的 condition。因此，混合类型的 condition 不共享一个全局调用参数顺序。

AnyOf 表达当前备选条件之间的竞速。如果严格优先级要求继续等待高优先级输入，请只返回该 condition，等它解决后再开放其他条件。

Channel 结果

class ChannelWaitStep(Step[int]):
    def __init__(self, approval: Channel[str]) -> None:
        self.approval = approval

    def wait_for(self, context: Context, input: int) -> Wait:
        return Wait.any_of(
            self.approval.for_one(),
            Timer.by_duration(timedelta(seconds=input)),
        )

    def execute(self, context: Context, input: int) -> StepDecision:
        if context.has_timer_fired():
            return graceful_complete("approval timed out")
        approvals = self.approval.results(context)
        return graceful_complete(approvals[0])

例子: examples/python/dex_examples/primitives/channel/channel_flow.py

SubFlow 结果

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}")

例子: examples/python/dex_examples/primitives/subflow/parent_flow.py

anyCombinationOf

anyCombinationOf 表示 AND 组的 OR。每组内用 all_of 组合构建。

各组合之间不会 greedy 地消费 Channel。Dex 只消费被选中组合里的 Channel condition。即使其他组合也已满足，只属于其他组合的 Channel condition 也不会消费消息，消息会继续留在队列中。

AnyCombinationOf 与普通等待组合器不同：组合中的每个 Condition 都需要唯一的 condition ID。

return Wait.any_combination_of(
    ConditionCombination.of(
        channel_a.for_one(condition_id="signal-a"),
        Timer.by_duration(timeout, condition_id="timeout"),
    ),
    ConditionCombination.of(
        channel_b.for_one(condition_id="signal-b"),
    ),
)

例子: examples/python/dex_examples/primitives/wait_types/wait_types_flow.py

Step execution local

SetStepExecutionLocal 把值存在当前 StepExecution 上，供 Execute 用 GetStepExecutionLocal 读取。适用于 WaitFor 算出了 Execute 需要的数据，又不想为此定义 Flow 级 Attribute。

注意： Step input 本身会传给两个方法，不要把 input 再拷进 local。

def wait_for(self, context: Context, input: int) -> Wait:
    context.set_step_execution_local("note", f"approval:{input}")
    return Wait.until(approval.for_one())

def execute(self, context: Context, input: int) -> StepDecision:
    note = context.get_step_execution_local("note", str)
    return graceful_complete(note or "")

例子: examples/python/dex_examples/primitives/step_execution_local/step_execution_local_flow.py

Skip immediately

一种特殊的 Wait 是 SkipWaitImmediately。它不等待任何 condition。当构建 dynamic WaitFor 方法时，某些分支不需要等待，可以使用它。

return Wait.skip_immediately()

例子: examples/python/dex_examples/primitives/flow/example_flow.py