编写聚合

聚合是做出决定的单位：它一次接收一条命令，查看自己的状态，决定发生哪些事件。本页以 eShop 的订单为例写一个聚合。背后的模型见决定与演化。

代码放在哪里

一个聚合跨越它所在上下文的两个模块：

模块	包含	由谁读取
`*-api`	命令和事件，普通的数据类	服务本身，以及发送它的命令或读取它的事件的每个对端
`*-domain`	根类及其状态类	只有服务本身

编译器在构建期读取领域模块，生成聚合的模型、它的绑定以及它那部分 Manifest。运行时不扫描任何东西。

编写命令

命令是 api 模块里的一个数据类。它承载调用方要求的内容，从不承载聚合是谁：ID 来自路由或发送命令的 Saga，租户、所有者和空间来自调用方，期望版本来自 If-Match。

kotlin

/**
 * Submits an order of [items]. The basket context sends it when its buyer checks out, with the order's id fixed by the
 * basket. The order belongs to the buyer: a saga's command carries the scope of the basket it is derived from, whose
 * owner is its buyer (decision D189), and the order takes its owner from the command that creates it.
 */
@CreateAggregate
@Summary("Submit an order")
data class SubmitOrder(
    val items: List<OrderItem>,
)

@Summary 为命令提供一行摘要，用于 Manifest 和 OpenAPI 文档。不带数据的命令是一个 data object，例如 ShipOrder。

命令能否创建它的聚合，由它的类决定：

命令类	何时被接受
`@CreateAggregate`	聚合尚不存在；否则 `wow.aggregate.exists`
`@AllowCreate`	无论聚合是否存在
两者都没有	聚合已存在；否则 `wow.aggregate.not-found`

每条命令的路由是 POST /<context>/<aggregate>/{id}/<command>，命令名采用 kebab case，与 Manifest 列出的路由一致：POST /eshop-ordering/order/{id}/submit-order。创建命令还提供不带 {id} 的路由，由服务端为聚合分配 ID。准入会先查找同一个请求 ID 已经创建过的聚合并复用它的 ID，所以同一个创建发两次只会创建一个聚合。

编写根

根是一个标注了 @AggregateRoot、由它的状态构造的类。它的命令函数做出决定：每个函数接收一条命令，返回它的决定。

kotlin

@OnCommand(produces = [OrderSubmitted::class])
fun onCommand(command: SubmitOrder): Decision {
    val invalid = command.items.isEmpty() || command.items.any { it.quantity <= 0 || it.unitPrice.signum() < 0 }
    if (invalid) {
        return Decision.reject(Rejections.INVALID_ITEMS, "The order's items are not valid: ${command.items}.")
    }
    return Decision.events(OrderSubmitted(command.items))
}

编译器找出每个标注 @OnCommand 或名为 onCommand 的非私有函数。第一个参数是命令；返回值是决定：

返回	含义
`Decision.events(e1, e2, …)`	这些事件，作为一条事件流一起追加
`Decision.NoOp`	命令被接受但什么都不改变；不写入任何东西
`Decision.reject(code, message)`	命令被拒绝；不写入任何东西，调用方拿到错误码（拒绝与错误码）
一个事件，或一个事件列表	这些事件
`Unit`、`null` 或空列表	什么都不改变

返回 Decision 的函数没有告诉编译器它能产生哪些事件，所以要在 @OnCommand(produces = [...]) 里列出它们。Manifest 会发布这份列表。返回类型本身已经指明事件的函数，例如返回某个事件类或它们的密封类型，不必重复列出。

聚合以根类名的 snake case 命名（Order 是 order，ViewPreferences 是 view_preferences），或者用 @Name 指定。

读取准入后的命令

需要准入从请求中得知的信息的命令函数，接收 Command<X> 而不是 X：

成员	包含
`body`	命令 `X`
`ref`	聚合：它的上下文、名字、租户和 ID
`scope`	调用方的所有者和空间
`headers`	宿主保留的请求事实（`wow.http.request-facts`），例如调用的应用
`causation`、`operator`	命令所属的链，以及它以谁的名义记录

购物车的 ID 就是买家的用户 ID，所以它的第一条命令用它来核对调用方：

kotlin

@OnCommand(produces = [BasketItemAdded::class])
fun onCommand(command: Command<AddBasketItem>): Decision {
    val owner = command.scope.ownerId
    if (owner != null && owner != command.ref.id) {
        return Decision.reject(Rejections.NOT_OWNER, "The basket ${command.ref.id} is not $owner's.")
    }
    val adding = command.body
    if (adding.quantity <= 0 || adding.unitPrice.signum() < 0) {
        return Decision.reject(Rejections.INVALID_ITEM, "$adding is not a valid item.")
    }
    return Decision.events(BasketItemAdded(adding.productId, adding.unitPrice, adding.quantity))
}

使用领域服务

命令之后的参数是领域服务，或者 DecideScope。服务会成为生成的模型的一个参数，宿主在构建模型时绑定一次：取应用中该类型唯一的 Bean，或者它的 primary Bean。在参数上写 @Name("…") 则取这个名字的 Bean，于是同一类型的两个参数可以取两个不同的 Bean。

订单把它的支付策略作为服务接收，并在库存确认后用 DecideScope 为自己定时取消：

kotlin

/**
 * Confirms one product. A confirmation the order does not wait for changes nothing: one read again from the
 * stock stream, or one that arrives after another product's shortage cancelled the order (that reservation is
 * released when the inventory hears of the cancellation).
 */
@OnCommand(
    produces = [OrderStockItemConfirmed::class, OrderStockConfirmed::class],
    schedules = [CANCEL_ORDER],
)
fun onCommand(command: ConfirmOrderStock, paymentPolicy: PaymentPolicy, scope: DecideScope): Decision {
    val waiting = state.status == OrderStatus.AWAITING_VALIDATION &&
        command.productId in state.products() &&
        command.productId !in state.confirmedProducts
    if (!waiting) {
        return Decision.NoOp
    }
    val confirmed = OrderStockItemConfirmed(command.productId)
    if (state.confirmedProducts + command.productId != state.products()) {
        return Decision.events(confirmed)
    }
    scope.schedule(
        PAYMENT_GRACE_PERIOD,
        ScheduleRule.after(paymentPolicy.gracePeriod),
        CancelOrder(PAYMENT_TIMEOUT)
    )
    return Decision.events(confirmed, OrderStockConfirmed)
}

服务可以做 I/O。会阻塞线程的函数，例如调用 JDBC 的函数，要标注 @Blocking，它会在虚拟线程上运行。函数也可以是 suspend，或者返回 Reactor 的 Mono，这需要类路径上有 wow-reactor。

向内核要更多东西

DecideScope 是一次决定除了事件之外还能向内核要的东西：

成员	作用
`command`	正在决定的命令，类型为 `Command<*>`
`ids`	节点的 ID 生成器，用于决定需要的 ID
`putResult(key, value)`	往事件追加之后给调用方的回答里加一个值。它不存储：重复的请求不会再得到它
`reserve(namespace, key)`	为这个聚合预留一个唯一键，例如用户名；被另一个聚合持有时返回 `false`
`release(namespace, key)`	在事件追加之后，释放这个聚合持有的一个键
`onAppendFailure { … }`	在确定事件没有追加时，撤销一个外部动作
`afterAppend { … }`	事件追加之后运行；失败只记日志，不抛出
`schedule(key, rule, command, target)`	定时一条命令，发往这个聚合，除非 `target` 指定了另一个。它与事件一起写入，要么都写入，要么都不写
`cancelSchedule(key)`	取消这个聚合在 `key` 下保持的定时

结账时用它提交的订单 ID 回答调用方：

kotlin

@Suppress("UnusedParameter") // The command has no body: its type is what the function handles.
@OnCommand(produces = [BasketCheckedOut::class])
fun onCommand(command: CheckoutBasket, scope: DecideScope): Decision {
    if (state.items.isEmpty()) {
        return Decision.reject(Rejections.EMPTY, "The basket is empty.")
    }
    val orderId = "${scope.command.ref.id}-${state.checkouts + 1}"
    scope.putResult(ORDER_ID_RESULT, orderId)
    return Decision.events(BasketCheckedOut(orderId, state.items))
}

调用方从结果的 results 里读取它：{"orderId": "alice-1"}。

定时命令的函数要在 @OnCommand(schedules = [...]) 里声明命令的完整类型名，就像订单对 eshop-ordering.order.CancelOrder 做的那样：编译器读不到函数体里的调用，而 Manifest 要列出它们。

编写状态

状态类为每个事件提供一个演化函数，标注 @OnSourcing 或名为 onSourcing。演化函数应用一个事件：它不接收服务，也不做 I/O。

kotlin

fun onSourcing(event: OrderSubmitted) {
    items = event.items
    totalAmount = event.items.fold(BigDecimal.ZERO) { total, item ->
        total + item.unitPrice.multiply(BigDecimal(item.quantity))
    }
    status = OrderStatus.SUBMITTED
}

聚合能产生的每个事件都需要一个演化函数，即使它什么都不改变。
原地修改状态的演化函数不返回任何东西。返回状态类型的演化函数则返回新的状态，用于不可变状态。
在第一个事件之前就给每个属性一个值。内核按状态的原样序列化它，用于快照和发布的事件流。
命令函数从不修改状态，只读取它。测试 DSL 比较每次决定前后的状态，若有改变就让测试失败。

在每条命令之后运行代码

标注 @AfterCommand 或名为 afterCommand 的函数，在它适用的命令函数之后运行：include 里的命令函数，include 为空时是全部，再减去 exclude 里的。它看到的是命令函数看到的那个状态，返回事件，这些事件在同一条流里追加在命令自己的事件之后。命令被拒绝时它不运行，它也不能拒绝。它失败时整条命令失败，什么都不写。

生命周期命令

每个聚合还处理 me.ahoo.wow.api.lifecycle 中的五条系统命令：DeleteAggregate、RestoreAggregate、TransferOwner、TransferSpace 和 ApplyTags。它们改变的是聚合的元数据，而不是它的状态；已删除的聚合拒绝除 RestoreAggregate 之外的每条命令。@AggregateRoot(lifecycle = [...]) 可以收窄这份列表；空列表则一条都不启用。

测试

用 aggregateSpec 测试每个命令函数：给定这些事件，执行这条命令，期望得到这些事件和这个状态。做法见测试领域代码。

下一步

拒绝与错误码：声明命令被拒绝时用的错误码。
编写 Saga：从聚合的决定派生命令。

编写聚合 ​

代码放在哪里 ​

编写命令 ​

编写根 ​

读取准入后的命令 ​

使用领域服务 ​

向内核要更多东西 ​

编写状态 ​

在每条命令之后运行代码 ​

生命周期命令 ​

测试 ​

下一步 ​

编写聚合

代码放在哪里

编写命令

编写根

读取准入后的命令

使用领域服务

向内核要更多东西

编写状态

在每条命令之后运行代码

生命周期命令

测试

下一步