Deferred Choice
查看 Markdown当 Flow 等待多个事件中的一个，并由最先发生的事件决定下一步时，使用 Deferred Choice。这些事件来自 Flow 外部，Flow 无法提前选定分支。
例如向客户发出一份报价。客户可以接受，也可以拒绝。如果 3 天内客户都没有回应，报价过期。最先发生的事件胜出，另外两个事件不会再改变已选的分支。
工作方式
一个 Step 同时等待所有事件。它的 WaitFor 返回 AnyOf，每个事件一个 Condition。每种客户回应是一个 Channel Condition，截止时间是一个 Timer。
第一个 Condition 满足后，这个 Step 的 Execute 运行。它检查是哪个 Condition 胜出，然后进入对应的分支。
rpc acceptCustomerOffer(note) -> OfferAccepted.publish(note)
rpc rejectCustomerOffer(note) -> OfferRejected.publish(note)
AwaitOfferResponse.waitFor -> AnyOf(OfferAccepted.ForOne(), OfferRejected.ForOne(), Timer(3 days))
AwaitOfferResponse.execute
if the Timer fired: goTo(ExpireOffer)
if OfferAccepted has a result: goTo(FulfillOffer)
otherwise: goTo(CloseRejectedOffer)
只有一个 Condition 会胜出，所以只有一个分支会开始。
规则
- AnyOf 只消费胜出 Condition 的消息。落败的回应仍留在它的 Channel 中等待。
- 多个 Channel Condition 同时就绪时，先声明的胜出。按业务优先级声明它们。在示例中，如果接受和拒绝都已在等待，选中的是接受。
- 在 Execute 中，用 HasTimerFired 检查 Timer，不要把“没有回应”当作已过期。读取 Channel 结果判断客户回应。落败的 Channel 没有结果。
- Timer 从 Step 开始等待时计时。Worker 和 Server 重启后它仍然有效。Step 等待期间不占用 Worker。
- 选择完成后才到达的回应不会改变已选的分支。Flow 仍在运行时，RPC 会成功，并发布一条没有 Step 消费的消息。Flow 关闭后，Dex 会拒绝这个 RPC。所以 RPC 调用成功并不代表这个回应胜出。调用方要知道最终结果，应读取 Flow 结果。
- 测试过期分支时，用 Client 的 SkipTimer 跳过 Timer。不要为了测试缩短业务截止时间。
相关形状
- Reminder、Inactivity tracker 和 Inbound Event Loop 也等待 Timer 和 Channel 的 AnyOf。它们会回到同一个 Step 循环。Deferred Choice 只决定一次，然后继续往下走。
- First-Win Parallel Steps 让 Flow 自己启动的工作互相竞争。Deferred Choice 让外部事件互相竞争。
示例
OfferResponseFlow 以报价 ID 启动。AwaitOfferResponse 等待 acceptCustomerOffer RPC、rejectCustomerOffer RPC 或 3 天的 Timer，然后进入 FulfillOffer、CloseRejectedOffer 或 ExpireOffer。这三个 Step 都用各自的结果完成 Flow。
# How long the customer has to respond.
OFFER_EXPIRY = timedelta(days=3)
OFFER_ACCEPTED = Channel("offerAccepted", str)
OFFER_REJECTED = Channel("offerRejected", str)
class AwaitOfferResponse(Step[str]):
"""Waits for whichever event happens first and routes on it."""
def wait_for(self, context: Context, offer_id: str) -> Wait:
return Wait.any_of(
OFFER_ACCEPTED.for_one(),
OFFER_REJECTED.for_one(),
Timer.by_duration(OFFER_EXPIRY),
)
def execute(self, context: Context, offer_id: str) -> StepDecision:
if context.has_timer_fired():
return go_to(ExpireOffer, offer_id)
if OFFER_ACCEPTED.results(context):
return go_to(FulfillOffer, offer_id)
return go_to(CloseRejectedOffer, offer_id)
例子: examples/python/dex_examples/patterns/deferred-choice/offer_response_flow.py
反模式
- 为每个事件启动一个并行 Step，先完成的取消其余 Step。一个 AnyOf 等待就能做出选择，分支之间没有竞争。
- 用 Timer 循环检查某个 Attribute，看客户是否已回应。
- 用单独的 Flow、cron 任务或数据库任务执行截止时间。