@abreza/nestjs-event-store 中文文档教程

Question

NestJs Event Store

NestJS CQRS 模块，支持 EventStore.org 和 NATS Streaming。 它需要@nestjs/cqrs。

<p align="center">

Installation

$ yarn add @juicycleff/nestjs-event-store
$ yarn add node-nats-streaming node-eventstore-client

Description

该模块旨在弥合 NestJs 与流行的事件存储代理（如 Event Store 和 ）之间的差距支持 kafka 的 NATS Streaming 即将到来。

Event Store

它支持 EventStore.Org 中所有不同的订阅策略， 例如 Volatile、CatchUp 和 Persistent 订阅相当容易。 支持用于存储追赶事件类型最后检查点位置的存储适配器接口，因此 检查点可以在启动时读取； 适配器接口非常简洁和简单，最好使用 EventStoreModule.registerFeatureAsync 方法进行分配。 即将推出适配器数据存储示例。

注意：如果您的事件代理类型是 Event Store，那么 featureStreamName 应该看起来像 '$ce-user'，那么您应该命名您的域参数应该是 user 而不是 $ce，例如。

export class UserCreatedEvent implements IEvent {
    constructor(
        public readonly user: any // This what im talking about.
    )  { }
}

其工作方式是我们根据构造函数名称中的第一个参数对事件进行分组，并且此参数名称必须是 featureStreamName 的子字符串。 很抱歉，您目前无法将您自己的唯一名称传递给您，但我会添加对它的

NATS Streaming

支持它支持所有具有共享订阅和易失性的持久/持久订阅。 它没有上述 Event Store 的限制。

注意：如果您的事件代理类型是 NATS，那么 featureStreamName 应该类似于 'user'。

Setup from versions from v3.1.15

Setup NATS

Setup root app module for NATS

import { Module } from '@nestjs/common';
import { EventStoreModule } from '@juicycleff/nestjs-event-store';

@Module({
  imports: [
    EventStoreModule.register({
      type: 'nats',
      groupId: 'groupId',
      clusterId: 'clusterId',
      clientId: 'clientId', // Optional (Auto generated with uuid)
      options: {
        url: 'nats://localhost:4222',
        reconnect: true,
        maxReconnectAttempts: -1,
      },
    }),
  ]
})
export class AppModule {}

Setup async root app module

import { Module } from '@nestjs/common';
import { EventStoreModule } from '@juicycleff/nestjs-event-store';
import { EventStoreConfigService } from './eventstore-config.service';

@Module({
  imports: [
    EventStoreModule.registerAsync({
      type: 'nats',
      useClass: EventStoreConfigService
    }),
  ]
})
export class AppModule {}

Setup feature module

import { Module } from '@nestjs/common';
import { UsersController } from './users.controller';
import { AccountEventHandlers, UserLoggedInEvent } from '@ultimatebackend/core';
import { AccountSagas } from '../common';
import { EventStoreModule, EventStoreSubscriptionType } from '@juicycleff/nestjs-event-store';

@Module({
  imports: [
    EventStoreModule.registerFeature({
      featureStreamName: 'user',
      type: 'nats',
      subscriptions: [
        {
          type: EventStoreSubscriptionType.Persistent,
          stream: 'account',
          durableName: 'svc-user',
        }
      ],
      eventHandlers: {
        UserLoggedInEvent: (data) => new UserLoggedInEvent(data),
      },
    })
  ],
  providers: [...AccountEventHandlers, AccountSagas],
  controllers: [UsersController],
})
export class UsersModule {}

Setup EventStore

Setup root app module for EventStore

import { Module } from '@nestjs/common';
import { EventStoreModule } from '@juicycleff/nestjs-event-store';

@Module({
  imports: [
    EventStoreModule.register({
      type: 'event-store',
      tcpEndpoint: {
        host: 'localhost',
        port: 1113,
      },
      options: {
        maxRetries: 1000, // Optional
        maxReconnections: 1000,  // Optional
        reconnectionDelay: 1000,  // Optional
        heartbeatInterval: 1000,  // Optional
        heartbeatTimeout: 1000,  // Optional
        defaultUserCredentials: {
          password: 'admin',
          username: 'chnageit',
        },
      },
    }),
  ]
})
export class AppModule {}

Setup async root app module

import { Module } from '@nestjs/common';
import { EventStoreModule } from '@juicycleff/nestjs-event-store';
import { EventStoreConfigService } from './eventstore-config.service';

@Module({
  imports: [
    EventStoreModule.registerAsync({
      type: 'event-store',
      useClass: EventStoreConfigService
    }),
  ]
})
export class AppModule {}

Setup feature module

import { Module } from '@nestjs/common';
import { CommandBus, CqrsModule, EventBus } from '@nestjs/cqrs';
import { EventStoreModule, EventStore, EventStoreSubscriptionType } from '@juicycleff/nestjs-event-store';

import {
  UserCommandHandlers,
  UserLoggedInEvent,
  UserEventHandlers,
  UserQueryHandlers,
} from '../cqrs';
import { UserSagas } from './sagas';
import { MongoStore } from './mongo-eventstore-adapter';

@Module({
  imports: [
    CqrsModule,
    EventStoreModule.registerFeature({
      featureStreamName: '$ce-user',
      type: 'event-store',
      store: MongoStore, // Optional mongo store for persisting catchup events position for microservices to mitigate failures. Must implement IAdapterStore
      subscriptions: [
        {
          type: EventStoreSubscriptionType.CatchUp,
          stream: '$ce-user',
          resolveLinkTos: true, // Default is true (Optional)
          lastCheckpoint: 13, // Default is 0 (Optional)
        },
      ],
      eventHandlers: {
        UserLoggedInEvent: (data) => new UserLoggedInEvent(data),
      },
    }),
  ],

  providers: [
    UserSagas,
    ...UserQueryHandlers,
    ...UserCommandHandlers,
    ...UserEventHandlers,
  ],
})
export class UserModule {}

Setup from versions from v3.0.0 to 3.0.5

Setup root app module

import { Module } from '@nestjs/common';
import { EventStoreModule } from '@juicycleff/nestjs-event-store';

@Module({
  imports: [
    EventStoreModule.register({
      tcpEndpoint: {
        host: process.env.ES_TCP_HOSTNAME || AppConfig.eventstore?.hostname,
        port: parseInt(process.env.ES_TCP_PORT, 10) || AppConfig.eventstore?.tcpPort,
      },
      options: {
        maxRetries: 1000, // Optional
        maxReconnections: 1000,  // Optional
        reconnectionDelay: 1000,  // Optional
        heartbeatInterval: 1000,  // Optional
        heartbeatTimeout: 1000,  // Optional
        defaultUserCredentials: {
          password: AppConfig.eventstore?.tcpPassword,
          username: AppConfig.eventstore?.tcpUsername,
        },
      },
    }),
  ]
})
export class AppModule {}

Setup async root app module

import { Module } from '@nestjs/common';
import { EventStoreModule } from '@juicycleff/nestjs-event-store';
import { EventStoreConfigService } from './eventstore-config.service';

@Module({
  imports: [
    EventStoreModule.registerAsync({
      useClass: EventStoreConfigService
    }),
  ]
})
export class AppModule {}

Setup module

注意 featureStreamName 字段并不重要，如果你的订阅类型是持久的'

Setup feature module

import { Module } from '@nestjs/common';
import { CommandBus, CqrsModule, EventBus } from '@nestjs/cqrs';
import { EventStoreModule, EventStore, EventStoreSubscriptionType } from '@juicycleff/nestjs-event-store';

import {
  UserCommandHandlers,
  UserCreatedEvent,
  UserEventHandlers,
  UserQueryHandlers,
} from '../cqrs';
import { UserSagas } from './sagas';
import { MongoStore } from './mongo-eventstore-adapter';

@Module({
  imports: [
    CqrsModule,
    EventStoreModule.registerFeature({
      featureStreamName: '$ce-user',
      store: MongoStore, // Optional mongo store for persisting catchup events position for microservices to mitigate failures. Must implement IAdapterStore
      subscriptions: [
        {
          type: EventStoreSubscriptionType.CatchUp,
          stream: '$ce-user',
          resolveLinkTos: true, // Default is true (Optional)
          lastCheckpoint: 13, // Default is 0 (Optional)
        },
        {
          type: EventStoreSubscriptionType.Volatile,
          stream: '$ce-user',
        },
        {
          type: EventStoreSubscriptionType.Persistent,
          stream: '$ce-user',
          persistentSubscriptionName: 'steamName',
          resolveLinkTos: true,  // Default is true (Optional)
        },
      ],
      eventHandlers: {
        UserLoggedInEvent: (data) => new UserLoggedInEvent(data),
        UserRegisteredEvent: (data) => new UserRegisteredEvent(data),
        EmailVerifiedEvent: (data) => new EmailVerifiedEvent(data),
      },
    }),
  ],

  providers: [
    UserSagas,
    ...UserQueryHandlers,
    ...UserCommandHandlers,
    ...UserEventHandlers,
  ],
})
export class UserModule {}

Setup async feature module

import { Module } from '@nestjs/common';
import { EventStoreModule } from '@juicycleff/nestjs-event-store';
import { EventStoreFeatureService } from './user-eventstore-feature.service';

@Module({
  imports: [
    EventStoreModule.registerFeatureAsync({
      useClass: EventStoreFeatureService
    }),
  ]
})
export class AppModule {}

Setup from versions below v2.0.0

Setup root app module

import { Module } from '@nestjs/common';
import { NestjsEventStoreModule } from '@juicycleff/nestjs-event-store';

@Module({
  imports: [
    NestjsEventStoreModule.forRoot({
      http: {
        port: parseInt(process.env.ES_HTTP_PORT, 10),
        protocol: process.env.ES_HTTP_PROTOCOL,
      },
      tcp: {
        credentials: {
          password: process.env.ES_TCP_PASSWORD,
          username: process.env.ES_TCP_USERNAME,
        },
        hostname: process.env.ES_TCP_HOSTNAME,
        port: parseInt(process.env.ES_TCP_PORT, 10),
        protocol: process.env.ES_TCP_PROTOCOL,
      },
    }),
  ]
})
export class AppModule {}

Setup module

import { Module } from '@nestjs/common';
import { CommandBus, CqrsModule, EventBus } from '@nestjs/cqrs';
import { NestjsEventStoreModule, EventStore } from '@juicycleff/nestjs-event-store';

import {
  UserCommandHandlers,
  UserCreatedEvent,
  UserEventHandlers,
  UserQueryHandlers,
} from '../cqrs';
import { UserSagas } from './sagas';

@Module({
  imports: [
    CqrsModule,
    NestjsEventStoreModule.forFeature({
      name: 'user',
      resolveLinkTos: false,
    }),
  ],

  providers: [
    UserSagas,
    ...UserQueryHandlers,
    ...UserCommandHandlers,
    ...UserEventHandlers,
  ],
})
export class UserModule {
  constructor(
    private readonly command$: CommandBus,
    private readonly event$: EventBus,
    private readonly eventStore: EventStore,
  ) {}

  onModuleInit(): any {
    this.eventStore.setEventHandlers(this.eventHandlers);
    this.eventStore.bridgeEventsTo((this.event$ as any).subject$);
    this.event$.publisher = this.eventStore;
  }

  eventHandlers = {
    UserCreatedEvent: (data) => new UserCreatedEvent(data),
  };
}

Notice

2.0.0 版本灵感来自 nestjs-eventstore

License

这个项目是 MIT licensed。