发布与消费事件流

上下文通过发布它的聚合的事件流，分享它内部发生的事情。其他上下文像读自己的事件流一样读它们：处理函数指明聚合，接收对端的事件类，并在自己的检查点里记录位置。本页发布 ordering 的订单，并在 payment 上下文中读取它。它与上下文之间其他通信方式的比较，见上下文之间。

上下文发布什么

每个聚合默认发布它的状态事件：每次提交一条记录，带着这次提交的事件和提交之后的状态。Manifest 把它们列为上下文的 published 事件流，上下文的 Kafka Sink 把它们写入主题 wow.<context>.<aggregate>.state。@AggregateRoot 可以按聚合改变这一点：

设置	效果
published = false	不发布这个聚合
publishState = false	发布它的事件，但不带状态，用于状态太大不便发送的情况
topic = "…"	另一个主题名

ordering 原样发布它的订单，主题是 wow.eshop-ordering.order.state。发布的事件流是上下文契约的一部分：它的事件和状态的字段就是对端读取的东西，Manifest 的兼容检查把删除字段或改变字段类型视为破坏性变更。

发布需要 Kafka：主题和信封见 Kafka。

读取另一个上下文的事件流

处理函数接收那个上下文 api 模块里的事件类和状态视图。服务的 Manifest 根据那个上下文的契约找出所属聚合；两个聚合共有的事件用 @OnEvent(aggregates = [<Context>Contract.<Aggregate>.QUALIFIED_NAME]) 指明：

@OnEvent
fun onOrderStockConfirmed(
    event: StateTypedEvent<OrderStateView, OrderStockConfirmed>
): CommandRequest<RequestPayment> = CommandRequest.of(
    RequestPayment(event.ref.id, event.state.totalAmount),
    event.ref.id
)

• 事件类。 OrderStockConfirmed 是 ordering api 中的类。payment 的领域模块依赖 eshop-ordering-api，它的 jar 带着 ordering 的契约（META-INF/wow/contract.json）；不需要其他配置。
• 状态。 OrderStateView 由 ordering api 根据契约生成：事件流带状态的每个聚合一个视图，名字是状态类名加 View。它包含状态的字段：ordering 要求的字段不可空，其余可空。api 里的类（OrderItem、OrderStatus）直接引用；ordering 领域内部的类型复制成视图里的嵌套类；自由结构的内容是 ObjectNode（任意值是 JsonNode）。视图里没有的字段会被忽略，所以 ordering 增加字段不会破坏读者。上游发布时不带状态（publishState = false）时，api 不为它生成视图，处理函数把状态声明为可空：对没有状态的事件流使用不可空的状态，节点启动时会失败。
• 访问属性。 每条发布记录的 access 是提交之后聚合的访问属性（属主与空间也在其中），即上游随事件流发布的访问属性，带不带状态都有；事件流未盖章时为 null。
• 聚合。 aggregates 中的所有聚合属于同一个上下文，写作 context.aggregate。
• 种类。 Saga、投影和反应器以同样的方式读取发布的事件流。

订阅方没有类的事件

接收 StateTypedEvent<S, E> 的处理函数，只会为是 E 的事件被调用。接收整个 StateEvent<S> 的处理函数，在事件的修订号就是 api 契约声明的那个修订号时，拿到的是 api 的类；否则拿到的是一个 PublishedEvent：它的 name（eshop-ordering.order.OrderShipped）、revision 和 JSON payload。所以针对旧版 api 构建的订阅方，在上游增加一个修订号后仍能继续工作，并自行决定怎样处理载荷。

处理函数接收的事件类，若类路径上的契约没有声明，节点不会启动。

从哪里开始

订阅另一个上下文事件流、还没有检查点的订阅，按它的种类开始：

• 投影从主题的开头开始，用之前发布的内容填满它的读模型。
• Saga 或反应器的处理函数从它的服务加入这条流的地方开始。服务里还没有别的订阅读过这个主题时，就是开头，所以之后才加入的服务仍能拿到之前发布的内容。已经有订阅读过时，就是末尾，所以给已在读这条流的服务后来新增的处理函数，不会再对它的历史采取行动：新的 Saga 处理函数不会为服务很久以前读过的事件发送命令。Manifest 把这种起点写作 JOIN。

@OnEvent(start = OnEvent.Start.BEGINNING) 或 END 固定它，宿主的 wow.subscriptions.[<context>.<processor.function>].start 覆盖两者，取值为 beginning、end 或一个时刻。时刻让订阅从每个分区中按偏移量顺序第一个提交时间不早于它的记录开始，此后的每条记录都会处理，无论其提交时间；没有这样的记录时从末尾开始。它在分区上的首次启动会从保留的第一条记录起读取来找到这条记录，跳过无法解码的记录。有了第一个检查点之后，它就从那里继续。

服务读取主题时不使用 Kafka 消费组：偏移量保存在它自己的检查点存储里，与它其他订阅的位置放在一起。

从 9.x 导入的历史

从 9.x 迁移过来的服务也会发布 migrate-data 导入到它日志里的历史，让对端的投影能用它构建读模型。导入记下的切换点及之前的每条记录都标为导入的（状态事件里的 imported）。Saga 或反应器像跳过它不处理的事件一样跳过这样的记录，因为 9.x 已经对它采取过行动；投影以及任何读取这个主题的系统都会处理它。从未迁移过的服务不会发布带这个标记的记录。

回应上游

对对端事件作出反应的 Saga，常常用一条发往那个对端的命令来回应：支付网关批准付款后，payment 为订单付款。

@OnEvent
fun onPaymentSucceeded(
    event: StateTypedEvent<PaymentState, PaymentSucceeded>
): CommandRequest<PayOrder> = CommandRequest.of(PayOrder(event.ref.id, event.state.amount), event.state.orderId)

CommandRequest.of(body, id) 从 ordering api jar 中的契约找到 ordering 的订单，命令发往 ordering 的命令主题 wow.eshop-ordering.commands。拒绝以回执的形式返回到 Saga 的 @OnRejected 处理函数：

/** An order that refuses its payment has the payment refunded. */
@OnRejected
fun onPayRejected(rejected: CommandReceipt.Rejected<PayOrder>): CommandRequest<RefundPayment> =
    CommandRequest.of(
        RefundPayment("The order ${rejected.ref.id} refused the payment: ${rejected.error.message}"),
        rejected.source.id,
    )

派生命令、回执和拒绝的完整说明，见编写 Saga。

外部消费者

不是 Wow 服务的系统读取同样的主题。每条记录都是一个以聚合为键的信封，所以一个聚合的记录保持顺序：见线上协议。

测试

测试用测试 DSL 的 stateEvent，以 api 的事件类和状态视图构建一次上游聚合的提交，交给处理函数：见测试领域代码。