---
title: "Inbound Event Loop"
description: "当 Flow 需要持续接收外部消息，直到收到结论或对话长时间没有动静时，使用这个模式。例如工单里的回复、聊天会话里的消息，或评审讨论里的答复。它是 Inactiveness Tracker Timer 的扩展：每条消息都会让不活跃 Timer 重新计时，另一个表示结论的消息会结束循环。"
canonical_url: "https://docs.superdurable.io/zh-Hans/design-patterns/inbound-event-loop/"
source: "docs/i18n/zh-Hans/docusaurus-plugin-content-docs/current/design-patterns/inbound-event-loop.mdx"
locale: "zh-Hans"
---

# Inbound Event Loop

当 Flow 需要持续接收外部消息，直到收到结论或对话长时间没有动静时，使用这个模式。例如工单里的回复、聊天会话里的消息，或评审讨论里的答复。它是 [Inactiveness Tracker Timer](https://docs.superdurable.io/zh-Hans/design-patterns/durable-timer/inactiveness-tracker-timer.md) 的扩展：每条消息都会让不活跃 Timer 重新计时，另一个表示结论的消息会结束循环。

## 结构

```text
rpc sendReply(text)   -> validate text; if invalid: return a rejection
                         Replies.publish(text); return accepted
rpc resolve(reason)   -> Resolutions.publish(reason)

AwaitInbound.waitFor  -> AnyOf(Replies.AtLeast(1), Resolutions.ForOne(), Timer(inactivity))
AwaitInbound.execute
    if the Timer fired:   goTo(CloseInactive)
    replies = the messages the reply Channel Condition consumed
    if replies is empty:  goTo(Resolve, the consumed resolution)
    Transcript = Transcript + replies
    goTo(AwaitInbound)    // a new execution of the same Step starts a fresh Timer
```

## 规则

- RPC 只校验输入，然后把它发布到 Channel。它不写入等待中的 Step 会读取或加锁的 Attribute。
- 只有一个等待中的 Step。它用 **AnyOf** 等待回复 Channel、结论 Channel 和不活跃 Timer。
- 先声明回复 Channel，再声明结论 Channel。多个 Channel Condition 同时就绪时，先声明的那个胜出，并且只消费胜出者的消息。所以在结论之前收到的回复总会先被记录。
- Channel Condition 只有胜出时才消费消息。**Execute** 运行期间到达的回复会留在队列里，由下一次执行处理，不会丢失。
- **Execute** 把消费到的消息追加到一个 Attribute，例如 transcript。等待中的 Step 是这个 Attribute 唯一的写入者。
- 然后 **Execute** 回到同一个 Step。新的执行会注册新的等待，所以每条消息之后不活跃 Timer 都会重新开始。
- 可以用至少一条的 Condition 一次取走所有排队的回复，也可以每次执行只取一条。
- 让 transcript 保持有界。长对话把消息存放在 [sequentially chunked AttributeMap](https://docs.superdurable.io/zh-Hans/design-patterns/sequentially-chunked-attribute-map.md) 中。

## 为什么不让 RPC 写 transcript

如果由 RPC 追加 transcript，等待中的 Step 就必须锁住 transcript，RPC 为了读取再修改也必须加锁。锁不会排队。只要 Step 或另一个 RPC 持有这把锁，新到的 RPC 都会因为锁冲突被拒绝，客户端只能重试。发布到 Channel 不需要锁，所以并发的回复不会冲突。

## 结论之后到达的回复

如果 Resolve Step 立即结束 Flow，那么在结论之后、Flow 关闭之前收到的回复不会被记录。需要保留每一条已接受的回复时，让 Resolve 在回复 Channel 上用 **ForceCompleteIfChannelsEmpty** 结束 Flow，做法见 [Drain External Channels](https://docs.superdurable.io/zh-Hans/design-patterns/draining-channel-for-external-publishing.md)。它的 fallback 带着结论回到等待中的 Step。这个 Step 记录剩下的回复，然后再次进入 Resolve。回复 Channel 只能由等待中的 Step 消费。

## 示例

这个模式还没有专门的可运行示例，之后会补上。目前最接近的可运行起点是 [Inactiveness Tracker Timer](https://docs.superdurable.io/zh-Hans/design-patterns/durable-timer/inactiveness-tracker-timer.md) 的示例，它展示了 Timer 与 Channel 的竞争，以及回到同一个 Step 的循环。在它的基础上加上结论 Channel、transcript Attribute 和 RPC 里的校验即可。

## 相关内容

- [Channel](https://docs.superdurable.io/zh-Hans/primitives/channel.md)
- [Timer](https://docs.superdurable.io/zh-Hans/primitives/timer.md)
- [RPC](https://docs.superdurable.io/zh-Hans/primitives/rpc.md)
- [Drain Channels](https://docs.superdurable.io/zh-Hans/design-patterns/drain-channels.md)