---
title: "Deferred Choice"
description: "当 Flow 等待多个事件中的一个，并由最先发生的事件决定下一步时，使用 Deferred Choice。这些事件来自 Flow 外部，Flow 无法提前选定分支。"
canonical_url: "https://docs.superdurable.io/zh-Hans/design-patterns/deferred-choice/"
source: "docs/i18n/zh-Hans/docusaurus-plugin-content-docs/current/design-patterns/deferred-choice.mdx"
locale: "zh-Hans"
---

# Deferred Choice

当 Flow 等待多个事件中的一个，并由最先发生的事件决定下一步时，使用 Deferred Choice。这些事件来自 Flow 外部，Flow 无法提前选定分支。

例如向客户发出一份报价。客户可以接受，也可以拒绝。如果 3 天内客户都没有回应，报价过期。最先发生的事件胜出，另外两个事件不会再改变已选的分支。

**Flow：OfferResponseFlow**

起始 Step：`AwaitOfferResponse`

**Step 列表**

- `AwaitOfferResponse`：Wait for an accept, a reject, or the offer to expire.
  - WaitFor `anyOf`：channel `offerAccepted.for 1`, channel `offerRejected.for 1`, timer `OFFER_EXPIRY timer`
- `CloseRejectedOffer`：Close the rejected offer.
  - `gracefulComplete`
- `ExpireOffer`：Close the offer that expired without a response.
  - `gracefulComplete`
- `FulfillOffer`：Fulfill the accepted offer.
  - `gracefulComplete`

**连接**

- `goTo` (decision; AwaitOfferResponse; 条件 `context.has_timer_fired()`) → `ExpireOffer` (step)：go_to
- `goTo` (decision; AwaitOfferResponse; 条件 `not (context.has_timer_fired()) and OFFER_ACCEPTED.results(context)`) → `FulfillOffer` (step)：go_to
- `goTo` (decision; AwaitOfferResponse; 条件 `not (context.has_timer_fired()) and not (OFFER_ACCEPTED.results(context))`) → `CloseRejectedOffer` (step)：go_to
- `offerAccepted` (channel) → `AwaitOfferResponse` (step)：results
- `offerAccepted` (channel) → `anyOf` (wait; AwaitOfferResponse)：offerAccepted.for 1
- `offerRejected` (channel) → `anyOf` (wait; AwaitOfferResponse)：offerRejected.for 1
- `accept_customer_offer` (rpc) → `offerAccepted` (channel)：publish
- `reject_customer_offer` (rpc) → `offerRejected` (channel)：publish

例子: [examples/python/dex_examples/patterns/deferred-choice/offer_response_flow.py](https://github.com/superdurable/dex/tree/main/examples/python/dex_examples/patterns/deferred-choice/offer_response_flow.py)

## 工作方式

一个 Step 同时等待所有事件。它的 **WaitFor** 返回 **AnyOf**，每个事件一个 Condition。每种客户回应是一个 Channel Condition，截止时间是一个 Timer。

第一个 Condition 满足后，这个 Step 的 **Execute** 运行。它检查是哪个 Condition 胜出，然后进入对应的分支。

```pseudocode
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](https://docs.superdurable.io/zh-Hans/design-patterns/durable-timer/reminder.md)、[Inactivity tracker](https://docs.superdurable.io/zh-Hans/design-patterns/durable-timer/inactiveness-tracker-timer.md) 和 [Inbound Event Loop](https://docs.superdurable.io/zh-Hans/design-patterns/inbound-event-loop.md) 也等待 Timer 和 Channel 的 **AnyOf**。它们会回到同一个 Step 循环。Deferred Choice 只决定一次，然后继续往下走。
- [First-Win Parallel Steps](https://docs.superdurable.io/zh-Hans/design-patterns/parallel/first-win.md) 让 Flow 自己启动的工作互相竞争。Deferred Choice 让外部事件互相竞争。

## 示例

**OfferResponseFlow** 以报价 ID 启动。**AwaitOfferResponse** 等待 **acceptCustomerOffer** RPC、**rejectCustomerOffer** RPC 或 3 天的 Timer，然后进入 **FulfillOffer**、**CloseRejectedOffer** 或 **ExpireOffer**。这三个 Step 都用各自的结果完成 Flow。

**Python**

```python
# 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](https://github.com/superdurable/dex/tree/main/examples/python/dex_examples/patterns/deferred-choice/offer_response_flow.py)

**Go**

```go
// OfferExpiry is how long the customer has to respond.
const OfferExpiry = 3 * 24 * time.Hour

var (
	OfferAccepted = dex.DefineChannel[string]("offerAccepted")
	OfferRejected = dex.DefineChannel[string]("offerRejected")
)

// AwaitOfferResponse waits for whichever event happens first and routes on it.
type AwaitOfferResponse struct{ dex.StepDefaults }

func (AwaitOfferResponse) WaitFor(dex.Context, string) (*dex.Wait, error) {
	return dex.AnyOf(
		OfferAccepted.ForOne(),
		OfferRejected.ForOne(),
		dex.Timer(OfferExpiry),
	), nil
}

func (AwaitOfferResponse) Execute(ctx dex.Context, offerID string) (*dex.StepDecision, error) {
	if ctx.HasTimerFired() {
		return dex.GoTo(ExpireOffer{}, offerID), nil
	}
	accepted, err := OfferAccepted.GetConditionResults(ctx)
	if err != nil {
		return nil, err
	}
	if len(accepted) > 0 {
		return dex.GoTo(FulfillOffer{}, offerID), nil
	}
	return dex.GoTo(CloseRejectedOffer{}, offerID), nil
}
```

例子: [examples/go/patterns/deferred-choice/workflow.go](https://github.com/superdurable/dex/tree/main/examples/go/patterns/deferred-choice/workflow.go)

**Java**

```java
/** How long the customer has to respond. */
public static final Duration OFFER_EXPIRY = Duration.ofDays(3);
public static final Channel<String> offerAccepted =
        Channel.define("offerAccepted", String.class);
public static final Channel<String> offerRejected =
        Channel.define("offerRejected", String.class);

/** Waits for whichever event happens first and routes on it. */
static final class AwaitOfferResponse implements Step<String> {
    @Override
    public Class<String> getInputType() {
        return String.class;
    }

    @Override
    public Wait waitFor(final Context context, final String offerId) {
        return Wait.anyOf(
                offerAccepted.forOne(),
                offerRejected.forOne(),
                Timer.byDuration(OFFER_EXPIRY));
    }

    @Override
    public StepDecision execute(final Context context, final String offerId) {
        if (context.hasTimerFired()) {
            return StepDecision.goTo(ExpireOffer.class, offerId);
        }
        if (!offerAccepted.getConditionResults(context).isEmpty()) {
            return StepDecision.goTo(FulfillOffer.class, offerId);
        }
        return StepDecision.goTo(CloseRejectedOffer.class, offerId);
    }
}
```

例子: [examples/java/src/main/java/io/superdurable/dex/patterns/deferredchoice/OfferResponseFlow.java](https://github.com/superdurable/dex/tree/main/examples/java/src/main/java/io/superdurable/dex/patterns/deferredchoice/OfferResponseFlow.java)

**TypeScript**

```typescript
/** How long the customer has to respond. */
export const OFFER_EXPIRY_MS = 3 * 24 * 60 * 60 * 1000;
export const offerAccepted = new Channel("offerAccepted", stringCodec);
export const offerRejected = new Channel("offerRejected", stringCodec);

/** Waits for whichever event happens first and routes on it. */
class AwaitOfferResponse implements Step<string> {
  public readonly inputCodec = stringCodec;

  public getStepType(): string {
    return "AwaitOfferResponse";
  }

  public waitFor(_context: Context, _offerId: string): Wait {
    return Wait.anyOf(
      offerAccepted.forOne(),
      offerRejected.forOne(),
      Timer.byDuration(OFFER_EXPIRY_MS),
    );
  }

  public execute(context: Context, offerId: string): StepDecision {
    if (context.hasTimerFired()) {
      return goTo(ExpireOffer, offerId);
    }
    if (offerAccepted.results(context).length > 0) {
      return goTo(FulfillOffer, offerId);
    }
    return goTo(CloseRejectedOffer, offerId);
  }
}
```

例子: [examples/typescript/src/patterns/deferred-choice/offer-response-flow.ts](https://github.com/superdurable/dex/tree/main/examples/typescript/src/patterns/deferred-choice/offer-response-flow.ts)

**Rust**

```rust
/// How long the customer has to respond.
pub const OFFER_EXPIRY: Duration = Duration::from_secs(3 * 24 * 60 * 60);
pub static OFFER_ACCEPTED: LazyLock<Channel<String>> =
    LazyLock::new(|| Channel::new("offerAccepted"));
pub static OFFER_REJECTED: LazyLock<Channel<String>> =
    LazyLock::new(|| Channel::new("offerRejected"));
pub const ACCEPT_CUSTOMER_OFFER: Rpc<String, ()> = Rpc::new("accept_customer_offer");
pub const REJECT_CUSTOMER_OFFER: Rpc<String, ()> = Rpc::new("reject_customer_offer");

/// Waits for whichever event happens first and routes on it.
#[derive(Default)]
pub struct AwaitOfferResponse;

impl Step for AwaitOfferResponse {
    type Input = String;

    fn wait_for(&self, _context: &mut Context, _offer_id: Self::Input) -> HandlerResult<Wait> {
        Ok(Wait::any_of([
            OFFER_ACCEPTED.for_one(),
            OFFER_REJECTED.for_one(),
            Timer::by_duration(OFFER_EXPIRY),
        ]))
    }

    fn execute(&self, context: &mut Context, offer_id: Self::Input) -> HandlerResult<StepDecision> {
        if context.has_any_timer_fired() {
            return Ok(StepDecision::go_to(&ExpireOffer, offer_id));
        }
        if !OFFER_ACCEPTED.condition_results(context)?.is_empty() {
            return Ok(StepDecision::go_to(&FulfillOffer, offer_id));
        }
        Ok(StepDecision::go_to(&CloseRejectedOffer, offer_id))
    }
}
```

例子: [examples/rust/src/control_patterns/deferred_choice/flow.rs](https://github.com/superdurable/dex/tree/main/examples/rust/src/control_patterns/deferred_choice/flow.rs)

## 反模式

- 为每个事件启动一个并行 Step，先完成的取消其余 Step。一个 **AnyOf** 等待就能做出选择，分支之间没有竞争。
- 用 Timer 循环检查某个 Attribute，看客户是否已回应。
- 用单独的 Flow、cron 任务或数据库任务执行截止时间。

## 相关内容

- [Timer](https://docs.superdurable.io/zh-Hans/concepts/primitives/timer.md)
- [Channel](https://docs.superdurable.io/zh-Hans/concepts/primitives/channel.md)
- [Durable Timer](https://docs.superdurable.io/zh-Hans/design-patterns/durable-timer.md)
- [Workflow Patterns 评估：WCP-16 Deferred Choice](https://docs.superdurable.io/zh-Hans/correctness-and-expressiveness/workflow-patterns.md#wcp-16)