Cell - CQRS 组件

@celljs/cqrs 将改变状态的代码（命令）与读取状态的代码（查询）分离，并让应用的其他部分对发生的事情做出反应（事件），方式参考 @nestjs/cqrs。处理器是 Cell 组件：通过装饰器查找，获得与其他组件相同的注入，在第一条消息分派时创建。除了 @celljs/core 外无其他依赖；仅限后端。

npm install @celljs/cqrs

命令与查询

命令或查询是类的实例；其类决定了唯一的处理器。扩展 CommandMessage<R> / QueryMessage<R> 来给 execute 指定结果类型（可选：没有它的话，execute<R>() 接受类型）。

import { Autowired } from '@celljs/core';
import { Body, Controller, Get, Param, Post } from '@celljs/mvc/lib/node';
import { CommandBus, CommandHandler, CommandMessage, QueryBus, QueryHandler, QueryMessage } from '@celljs/cqrs';

export class CreateUser extends CommandMessage<string> {
    constructor(readonly name: string) {
        super();
    }
}

export class GetUser extends QueryMessage<User | undefined> {
    constructor(readonly id: string) {
        super();
    }
}

@CommandHandler(CreateUser)
export class CreateUserHandler implements CommandHandler<CreateUser> {
    @Autowired(UserRepository)
    protected readonly users: UserRepository;

    execute(command: CreateUser): Promise<string> {
        return this.users.insert(command.name);
    }
}

@QueryHandler(GetUser)
export class GetUserHandler implements QueryHandler<GetUser> {
    @Autowired(UserRepository)
    protected readonly users: UserRepository;

    execute(query: GetUser): Promise<User | undefined> {
        return this.users.findById(query.id);
    }
}

@Controller('users')
export class UserController {
    @Autowired(CommandBus)
    protected readonly commandBus: CommandBus;

    @Autowired(QueryBus)
    protected readonly queryBus: QueryBus;

    @Post()
    create(@Body() body: { name: string }): Promise<string> {
        return this.commandBus.execute(new CreateUser(body.name));
    }

    @Get(':id')
    get(@Param('id') id: string): Promise<User | undefined> {
        return this.queryBus.execute(new GetUser(id));
    }
}

execute 解析为处理器返回的值，拒绝为其抛出的错误。装饰器接受 @Component 的选项作为第二参数（scope、proxy 用于 AOP 等）。

事件与传奇

一个事件类可以有任意数量的 @EventHandler 和 @Saga。传奇将事件转化为随之而来的命令：事件总线使用命令总线按顺序运行它返回的内容（一个命令、命令数组或无）。

import { EventBus, EventHandler, Saga } from '@celljs/cqrs';

export class UserCreated {
    constructor(readonly id: string, readonly name: string) { }
}

@EventHandler(UserCreated)                    // 或 @EventHandler([UserCreated, UserRenamed])
export class UserProjection implements EventHandler<UserCreated> {
    async handle(event: UserCreated): Promise<void> { ... }
}

@Saga(UserCreated)
export class WelcomeSaga implements Saga<UserCreated> {
    handle(event: UserCreated): SendWelcomeMail {
        return new SendWelcomeMail(event.id);
    }
}

await eventBus.publish(new UserCreated(id, name));

publish(event) 并发运行事件的处理器和传奇，在全部完成时解析，因此无服务器函数不会在它们之前返回；publishAll(events) 一个接一个地发布事件。处理器或传奇抛出的错误不会拒绝 publish 也不会阻止其他处理器：它会被记录并交给 UnhandledExceptionBus：

@Component(ApplicationLifecycle)
export class CqrsAlerts implements ApplicationLifecycle<Application> {
    @Autowired(UnhandledExceptionBus)
    protected readonly exceptions: UnhandledExceptionBus;

    initialize(): void {
        this.exceptions.onException(({ error, event, handler }) => { ... });
    }
}

聚合根

AggregateRoot 将其更改记录为事件：apply(event) 记录事件并调用聚合根的 on<EventClassName> 方法（如果存在），以更改状态；commit() 用事件总线发布记录的事件。先用 EventPublisher 将聚合根连接到事件总线。

import { AggregateRoot, EventPublisher } from '@celljs/cqrs';

export class User extends AggregateRoot {
    name: string;

    constructor(readonly id: string) {
        super();
    }

    rename(name: string): void {
        this.apply(new UserRenamed(this.id, name));
    }

    onUserRenamed(event: UserRenamed): void {
        this.name = event.name;
    }
}

@CommandHandler(RenameUser)
export class RenameUserHandler implements CommandHandler<RenameUser> {
    @Autowired(EventPublisher)
    protected readonly publisher: EventPublisher;

    @Autowired(UserRepository)
    protected readonly users: UserRepository;

    async execute(command: RenameUser): Promise<void> {
        const user = this.publisher.mergeObjectContext(await this.users.load(command.id));
        user.rename(command.name);
        await this.users.save(user);
        await user.commit();
    }
}

• loadFromHistory(events) 从存储的事件重建状态，不再次记录它们。
• getUncommittedEvents() 返回记录的事件；uncommit() 忘记它们。
• eventPublisher.mergeClassContext(User) 返回创建时连接实例的子类。
• 记录的事件不是聚合根的属性：JSON.stringify(user) 仅给出其状态。

错误

总线和 commit() 拒绝时返回 CqrsError，其 code 稳定，其 hint 说明如何修复：

Code	When
CELL_CQRS_NO_HANDLER	命令或查询的类没有 @CommandHandler / @QueryHandler
CELL_CQRS_DUPLICATE_HANDLER	一个命令或查询类有两个处理器
CELL_CQRS_NOT_AN_INSTANCE	消息是普通对象、null 或原始值，而不是类的实例
CELL_CQRS_AGGREGATE_NOT_MERGED	对 EventPublisher 未连接的聚合根调用 commit()；事件保持记录状态

自定义

CommandBus、QueryBus、EventBus、EventPublisher 和 UnhandledExceptionBus 是组件：用 rebind: true 替换其中之一，例如要同时将事件发送给消息代理：

@Component({ id: EventBus, rebind: true })
export class BrokerEventBus extends EventBusImpl {
    override async publish(event: object): Promise<void> {
        await super.publish(event);
        await this.broker.send(event);
    }
}

从 @nestjs/cqrs 迁移

@nestjs/cqrs	@celljs/cqrs
CqrsModule.forRoot()	添加依赖；模块随后端加载
ICommandHandler、IQueryHandler、IEventHandler	CommandHandler、QueryHandler、EventHandler：装饰器和接口用一个名字
@EventsHandler(A, B)	@EventHandler([A, B])
Command<R>、Query<R>	CommandMessage<R>、QueryMessage<R>（@celljs/mvc 有 @Query 装饰器）
@Saga() 带 RxJS 流	@Saga(Event) 类，其 handle(event) 返回命令；无 RxJS
eventBus.publish() 立即返回	await eventBus.publish() 等待处理器和传奇
aggregate.commit()	await aggregate.commit()；未连接时拒绝，而不是什么都不做
autoCommit	不支持：调用 commit()
UnhandledExceptionBus（RxJS subject）	UnhandledExceptionBus.onException（Cell Event）