英语 · 法语

@ddd-ts-kit / nats

NATS JetStream的集成事件提供者：an EventBus 的内核 到流并等待其确认，并将事件id作为流的消息id 丢弃中继重试，并从每个消费者和事件类型的一个持久拉消费者中消费 有明确的确认，有限数量的延迟重新交付和一封死信 流。的策略表中的“队列代理”行在JetStream中是等价的 docs/ARCHITECTURE.md，它就过去了 eventBusContract 的 @ddd-ts-kit/testing.

npm i @ddd-ts-kit/nats @ddd-ts-kit/events @nats-io/transport-node @nats-io/jetstream

测试了 @nats-io/transport-node 和 @nats-io/jetstream 3.4针对NATS Server 2.15 开始时 -js，在节点24上。核心NATS没有持久性：只有JetStream适合端口。

客户端

NATS项目模块化客户端： @nats-io/transport-node 对于连接， @nats-io/jetstream 对于流和消费者。两者都是3.4.0（发布于2026-05-13）， 纯JavaScript， 1.7 MB安装与他们的三个依赖，不安装脚本，和 每周下载约40万至30万次。遗留 nats 包（2.29）仍然 接收修复，但它是客户端的前一行；没有必要这么做。

作文根

处理程序永远不会看到这个包。他们发布到发件箱；中继发布到NATS； 每个消费上下文都以自己的名称通过总线侦听。这辆公共汽车是给 应用程序作为内核的两个端口： IntegrationEventPublisher 保留发件箱，并且 EventSubscribers 为每个上下文提供其自身消费者名称的订阅者。

init --broker nats 在项目的平台中编写此布线：由 platform/events/create-event-provider.ts， platform/composition/platform.module.ts 重新绑定 EventSubscribers 在上面，还有 main.ts 追踪到发件箱中继。用手， 同样的内容是：

// platform/messaging.module.ts
import { defineContextModule } from '@ddd-ts-kit/di';
import { EventSubscribers, IntegrationEventPublisher, type Logger, type Metrics } from '@ddd-ts-kit/kernel';
import { OutboxPublisher, OutboxStore } from '@ddd-ts-kit/events';
import { NatsEventBus } from '@ddd-ts-kit/nats';

declare const config: { readonly natsUrl: string };
declare const logger: Logger;
declare const metrics: Metrics;

export const bus = new NatsEventBus({
  connection: config.natsUrl, // or a list, or options of the client, or a factory opening a new connection
  stream: 'integration-events', // the default
  subjectPrefix: 'events', // the default: events go to events.<event type>
  replicas: 3, // default 1
  retry: { maxAttempts: 5, delayMs: 5_000 }, // the defaults
  prefetch: 10, // the default
  onFailure: (failure) => logger.warn('event bus failure', { ...failure }), // codes only, never an error message
  // The label `outcome` of the kernel is success or failure: the four outcomes of a delivery take another name.
  onDelivery: ({ consumer, outcome }) => metrics.counter('event_deliveries_total').increment({ consumer, status: outcome }),
});

export const messagingModule = defineContextModule({
  name: 'messaging',
  register(builder) {
    // Publishing side: handlers write to the outbox, in the transaction of the command.
    builder.rebind(IntegrationEventPublisher).toClass(OutboxPublisher, [OutboxStore]);
    // Consuming side: one subscriber per consuming context, named after it.
    builder.rebind(EventSubscribers).toValue({ forContext: (name) => bus.forConsumer(name) });
  },
});

import { Command, CommandBus, systemActor, type Actor, type IntegrationEvent } from '@ddd-ts-kit/kernel';
import { IntegrationEventHandler } from '@ddd-ts-kit/kernel/declarations';
import { defineIntegrationEvent, type PayloadOf } from '@ddd-ts-kit/events';
import { zodSchema } from '@ddd-ts-kit/validation/zod';
import { z } from 'zod';

// contexts/ordering/published-language: the contract billing listens to.
export const orderPlaced = defineIntegrationEvent({
  type: 'ordering.order.placed',
  version: 1,
  payload: zodSchema(z.object({ orderId: z.string() })),
});

// contexts/billing/application/commands/issue-invoice/issue-invoice.command.ts
export class IssueInvoice extends Command {
  constructor(
    readonly actor: Actor,
    readonly orderId: string,
  ) {
    super();
  }
}

// contexts/billing/application/integration/handlers/invoice-on-order-placed.integration-handler.ts
@IntegrationEventHandler(orderPlaced, { consumer: 'billing.invoice-on-order-placed' })
export class InvoiceOnOrderPlacedIntegrationHandler {
  constructor(private readonly commands: CommandBus) {}

  // Nobody is calling: the handler acts as a system actor named after its consumer.
  readonly actor = systemActor('billing.invoice-on-order-placed');

  async handle(event: IntegrationEvent<PayloadOf<typeof orderPlaced>>): Promise<void> {
    await this.commands.execute(new IssueInvoice(this.actor, event.payload.orderId));
  }
}

// main.ts
import { contextModule, kernelModule, type ContextModule, type DeclaredSource } from '@ddd-ts-kit/di';
import { createApplication } from '@ddd-ts-kit/di/tsyringe';
import { Clock } from '@ddd-ts-kit/kernel';
import { OutboxRelay, OutboxStore } from '@ddd-ts-kit/events';
import type { NatsEventBus } from '@ddd-ts-kit/nats';

// In a project: import { bus, messagingModule } from '@/platform/messaging.module';
declare const bus: NatsEventBus;
declare const messagingModule: ContextModule;
declare const persistenceModule: ContextModule;
declare const orderingDeclarations: DeclaredSource;
declare const billingDeclarations: DeclaredSource;

const app = createApplication({
  modules: [
    kernelModule(),
    persistenceModule,
    messagingModule,
    contextModule('ordering', orderingDeclarations),
    contextModule('billing', billingDeclarations),
  ],
});

// The relay target is the bus. A record is marked published when the stream acknowledged it.
const relay = new OutboxRelay({ store: app.resolve(OutboxStore), publisher: bus, clock: app.resolve(Clock) });

await app.start(); // subscribes the handler of billing on bus.forConsumer('billing')
await bus.assertTopology(); // fail at startup, not at the first message
relay.start({ intervalMs: 500 });

process.on('SIGTERM', async () => {
  await relay.stop(); // the tick in progress finishes
  await app.stop(); // the handlers are unsubscribed
  await bus.close(); // consumers stop, handlers in progress finish and acknowledge
});

• persistenceModule 结合 OutboxStore，并重新结合 UnitOfWork 和 ProcessedEventStore 到数据库：请参阅持久性适配器的README。声明的集成事件 处理程序在其自身上下文的订阅者(forContext 接收名称 的模块)，以及since UnitOfWork 和 ProcessedEventStore 是束缚的，它的幂等 标记和它的工作在一个工作单元中提交：一次重新交付是重复数据删除，一次失败 没有留下任何东西。
• 如果没有发件箱，总线就是内核模块的事件总线，处理程序发布 直接给经纪人： kernelModule({ eventBus: bus, eventSubscribers: () => ({ forContext: (name) => bus.forConsumer(name) }) }).
• 如果没有容器，同样的对象是手工连接的： bus.forConsumer('billing') 是一个 IntegrationEventSubscriber给予，给予 subscribeAll 和 consume 的 @ddd-ts-kit/events. 一辆没有 consumer 选项没有名称给它的持久消费者，所以 bus.subscribe(...) 抛出 ConsumerNameRequiredError：改为命名消费上下文。

import { consume, defineIntegrationEvent, subscribeAll } from '@ddd-ts-kit/events';
import type { ProcessedEventStore, UnitOfWork } from '@ddd-ts-kit/kernel';
import { NatsEventBus } from '@ddd-ts-kit/nats';
import { zodSchema } from '@ddd-ts-kit/validation/zod';
import { z } from 'zod';

declare const processed: ProcessedEventStore; // the store of the persistence adapter
declare const unitOfWork: UnitOfWork; // the transaction of the database
declare function issueInvoice(orderId: string): Promise<void>;

const orderPlaced = defineIntegrationEvent({
  type: 'ordering.order.placed',
  version: 1,
  payload: zodSchema(z.object({ orderId: z.string() })),
});

const bus = new NatsEventBus({ connection: 'nats://127.0.0.1:4222' });

// Instead of: bus.subscribe(orderPlaced.type, handler), which throws ConsumerNameRequiredError here.
// Write: the subscriber of one consuming context, whose name its durable consumers carry.
export const billing = subscribeAll(bus.forConsumer('billing'), [
  consume(orderPlaced, { consumer: 'billing.invoice-on-order-placed', processed, unitOfWork }, async (event) => {
    await issueInvoice(event.payload.orderId);
  }),
]);

await bus.assertTopology(); // creates billing_ordering_order_placed, then consumes

• publish(events) 只有当流被确认时才解析 每一个 批消息 作为存储，否则拒绝(PUBLISH_NOT_ACKNOWLEDGED， BROKER_UNAVAILABLE， EVENT_BUS_CLOSED）. 以后不会缓冲任何东西：发件箱就是缓冲区。
• 每条消息都带有报头 Nats-Msg-Id =事件id。溪水下降了一秒 在其重复窗口中具有相同id的消息(duplicateWindowMs， 两分钟 默认情况下，当流创建时设置)并且仍然确认它：一个中继 重试确认丢失的发布不会创建第二条消息。 该窗口覆盖了重试，而不是重播：消费者仍然重复数据删除。
• 订阅时必须提供消费者姓名： bus.forConsumer('billing') 给出了一个 IntegrationEventSubscriber 对于共享连接（模块化单体）上的一个上下文 有几个)，和 consumer 选项名称 bus.subscribe 为某项服务工作 这是一个单一的上下文)。
• 跟踪是从外部在两侧进行的，与任何提供程序一样： tracedPublisher(bus) 作为中继目标，在消费端对内核模块进行检测 （kernelModule({ instrumentation: traceApplication() }))，它给出每个声明的处理程序 一个跨度 process <consumer> 每一个尝试。 tracedSubscriber(bus.forConsumer(name)) 是为 没有的作文 kernelModule． 公共汽车载有 traceparent 和 correlationId 在信封和邮件头里都没动过。
• 这种相关性是双向的。通过发布 scopedPublisher(publisher, scope) 的 内核：声明的处理程序(或 consume(definition, { ..., scope }, handler) 用手, 包装的回调函数 inEventScope)在执行上下文中运行，其 correlationId 是发布的请求之一，是谁的 causationId 是事件， 再送一次也是在同一个地方。
• 提取到其自己的服务的上下文保留其消费者名称，因此它是持久的 消费者：服务器上没有任何变化。

拓扑结构

bus.describeTopology() 将流和持久消费者作为数据返回，以供查看 或生成的文档； bus.assertTopology() 幂等地创造它们。公共汽车也 在每个新连接上创建它们。为 forConsumer('billing') 订阅 ordering.order.placed：

持久名称既不包含点也不包含通配符：消费者名称和of的每个字符 外面的类型 [A-Za-z0-9_-] 变成了 _.

消息是JSON格式的信封(id， type， version， occurredAt， payload， traceparent， correlationId， tenantId)和头文件 Nats-Msg-Id （事件id）， x-event-type， x-event-version， traceparent， x-correlation-id 和 x-tenant-id． 任何NATS 客户端可以使用它。

一种类型的两个消费者各有一个持久消费者，并且都接收每个事件。两个 一个消费者的实例从同一个持久消费者中提取：每个事件都进入其中一个。

已存在的流永远不会被更改：总线接受同名的流，无论其名称是什么 限制，所以保留，最长时间和大小是你自己设定的。现存的持久的 消费者接受 ack_wait 和 max_deliver 声明它的公共汽车。

订购

无。流保持发布顺序，但由消费者处理 prefetch 消息 一旦相互竞争的实例相互交错，并且在消息被发送后返回重新发送 紧随其后。在有效负载中携带一个版本，或者在订单时使用Kafka提供程序 总很重要。

重试和死信行为

1. 消息在每个处理程序向其持久消费者订阅后得到确认 解决。一个声明 @IntegrationEventHandler (或 consume 的 @ddd-ts-kit/events 由 手)提交它的工作和它的幂等标记在一个单位的工作：确认 然后在提交之后，重新交付被重复数据删除。
2. 当处理程序抛出时，总线会延迟(nak 与 retry.delayMs)：服务器在该延迟之后再次发送它到任何实例。这个 尝试数是服务器的交付计数。没有热循环。
3. 在 retry.maxAttempts失败的投递，公共汽车给死信发了一份副本 主题, x-dead-reason: max-attempts， x-error-code， x-attempt 和 x-source （<stream>/<sequence>)，等待死信流的确认，然后 终止原(term）. 如果不能存储该副本，则拒绝该消息 用延误代替，并停放在晚些时候交付。
4. 消息体不是信封的消息(无法解析，没有。 id， type， version 或者 occurredAt，或者用 tenantId 那 isTenantId （内核的拒绝）是毒药：它 直接进入死信流 x-dead-reason: poison 和 x-error-code: POISON_MESSAGE，没有处理程序看到它。
5. 只有代码传播： x-error-code 是 code 错误或其类名的。的消息 永远不会记录、报告或附加错误，因为它可能引用有效负载数据。

为什么是明确的副本，而不是建议。 当消息到达 max_deliver 这个 服务器停止发送并发出 $JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES． 那 advisory是NATS的核心信息：如果没有人在那一刻听，它就会丢失，它会携带一条流 序列和没有故障代码，并且消息本身留在没有消费者的流中 我会再读一遍的。总线知道失败代码，并且知道这是最后一次尝试，所以它 将消息本身放置在一个公认的出版物上。

为什么 max_deliver 是两倍 maxAttempts. 当一次交付时，交付数量也会增加 在没有判决的情况下结束：进程在中间处理程序中被杀死，确认等待已过期， 连接切断。与 max_deliver 等于 maxAttempts，这样的交付可以使用 最后一个，然后服务器将静默地从消费者那里丢弃消息。公共汽车 公园在 maxAttempts;服务器限制是高于此限制的支撑。一个耗尽的信息 即使是这样，它也停留在流中，没有交付：上面的建议是值得警惕的。

交付至少一次。复制-终止命令意味着交付之间的崩溃 消息再次保存并停放它：死信流保留一个副本，因为该副本 有自己的消息id (<durable>.<sequence>）。

在失去连接时，客户端自己重新连接，它的pull消费者恢复：流 持久消费者驻留在服务器上，不需要再次声明任何内容。而服务器 不在 publish 立即拒绝 BROKER_UNAVAILABLE以及一份正在飞行中的出版物 拒绝后 publishTimeoutMs （默认为5秒） PUBLISH_NOT_ACKNOWLEDGED，所以 发件箱中继保存记录并重试它们。正在处理的消息为 之后再次交付 ackWaitMs． 总线自己打开第一个连接 指数回退(reconnect（200毫秒增加到30秒），如果是客户端，则需要一个新的时间 永不放弃。

凭证。 的URL或选项 connection 由打开连接的函数持有，不包含任何字段。一个 NatsEventBus 这是记录，传播或 连载显示 NatsEventBus [connection concealed] 没有别的了：它的域是不可数的， 和 toJSON 和 util.inspect 回答那个名字。它报告的失败是代码。什么 客户端库本身显示，当它自己的对象之一被检查时，不在 这个包：不记录那些。

处理器被告知的内容。 在事件旁边，每个处理程序接收传递 （EventDelivery 的内核)： { attempt, maxAttempts, system: 'nats', destination, consumer }， 哪里 destination 是主题和 consumer 耐久消费者 （billing_ordering_order_placed）. 应用程序的处理程序忽略它； @ddd-ts-kit/tracing 读它 每次尝试创建一个CONSUMER span，并标记重新交付和之后的尝试 事件被掩埋。公共汽车自己命名 messagingSystem: 'nats', traceparent 旅行在 并将信封作为真实的报头，完好无损地跨一次重试。

操作

看点：

有了错误跟踪器， DeadLetterReports 核的旋转数 onFailure 的报告 被放弃的事情，而且仅仅是那些： onFailure: (failure) => reports.onFailure(failure), reports.observing(bus.forConsumer(name)) 作为每一个上下文的订阅者，使一个报告 携带处理程序抛出的内容。 init 当项目有一个时连接它。

重播死消息，一旦原因被修复：

import type { NatsEventBus } from '@ddd-ts-kit/nats';

declare const bus: NatsEventBus;

// Up to 100 messages billing gave up on for this type, oldest first.
const replayed = await bus.replayDead({ consumer: 'billing', type: 'ordering.order.placed' }, 100);
console.log(`${String(replayed)} message(s) replayed`);

它将该消费者的许多消息从死信流移回 他们来自的对象，最老的一个，每个人都有一份被确认的副本，然后被删除 来自死信流，并带有头 x-replay-for 命名持久消费者。 此套件的其他使用者会确认并跳过为其他人标记的消息；一个 未使用此工具包编写的消费者将第二次看到事件，并且必须重复数据删除。这个 Copy是一条新消息：它的发送计数重新开始。看看 x-dead-reason 和 x-error-code 首先，重播有毒信息只会再次杀死它。

要永久地停止一个消费者，请删除它的持久消费者：取消订阅会留下它们，并且 流与 interest 保留将永远保留他们的信息。

限制

• 没有订购，见上文。
• 一个延迟。 每次尝试的等待都是一样的。这个 backoff JetStream的列表 没有使用Consumer：它适用于过期的确认等待，而不是拒绝。
• 处理程序必须在 ackWaitMs （默认30秒）。除此之外，服务器 在第一个实例仍在运行时将消息传递给另一个实例。公共汽车不发。 进行中信号：raise ackWaitMs 在最坏的处理者之上。
• 副本窗口很短 并属于溪流：两分钟除非溪流 以其他方式创建，并且现有流保留自己的流。它吸收一个中继重试，而不是 一个小时后，一张唱片从已死的发件箱中复活了。
• 保留是 limits，没有限制：事件流会增长，直到您设置最大值 年龄或尺寸。不要切换到 workqueue两个消费者不可能同时得到 每一个事件。与 interest，每一个耐用消费者必须先于第一个存在 它关心的出版物。
• 在发布接收到它后创建的消费者 (提供政策 all，只要… 流保留了它：一个新的消费上下文重播其类型保留的历史。
• 重新交付将再次运行持久消费者的每个处理程序，那些成功的 包括，当一个使用者的多个处理程序订阅相同类型时。每个都必须是 幂等。
• 批处理不是原子的。 当 publish 不合格品，一些事件的批次可能已经 存储。中继逐个发布并重试；流重复数据删除，然后是消费者。
• 没有消费者收听的事件将被存储没有等价物 onUnroutable.
• 两个仅由替换字符不同的名称发生冲突： billing.eu 和 billing_eu 都是同样的耐用消费者。
• replayDead 直接读取死信流，这需要 allow_direct : 总线将它设置在它创建的流上；把它放在你自己创建的一个上。
• 连接工厂必须在每次调用时打开一个新连接 应该保持 客户端重新连接无限制(maxReconnectAttempts: -1这是一辆公共汽车 给出一个URL)；公共汽车开什么关什么。
• 在服务器不在时关闭不会等待正在发送的确认 已发送：这些消息被再次发送。

未被测试套件验证：由多个服务器组成的集群(1以上的副本，leader 选举（失去quorum的流）、TLS和身份验证、确认等待 在处理程序期间过期，消费者进程在处理程序中被终止 MAX_DELIVERIES 超出总线限制的路径、持久消费者或在总线运行时删除的流(总线 然后重新打开连接，重新声明一切；路径没有被执行)， interest 保留，2.15以前的服务器，以及WebSocket传输。

其他供应商

@ddd-ts-kit/rabbitmq 和 @ddd-ts-kit/kafka 实现相同的两个端口并传递相同的端口 合同套件。这三个共享信封，标题名称(traceparent， x-correlation-id， x-event-version， x-attempt， x-error-code， x-dead-reason), 选项名称(consumer， retry， reconnect， onFailure， onDelivery)和方法 （forConsumer， assertTopology， describeTopology， replayDead， close）。

测试

每个套件都在单元运行中运行：总线通过真正的客户端和它的TCP连接 传输到NATS服务器，测试在一个空闲端口上以自己的进程开始 （test/support/server/）. 它使用核心协议并回答总线对JetStream的要求 正如nat -server 2.15所做的那样：限制流和它们的重复窗口，持久地拉动消费者 明确的确认, ack_wait， max_deliver，一个延迟的 -NAK， +TERM，直接得到 错误10058、10059、10148和10026。与 DDD_TS_KIT_TEST_NATS_URL，同样的套房经营a 第二次是针对真实服务器，这是为了保持内存中的一个诚实：

docker run -d --rm --name ddd-ts-kit-test-nats -p 127.0.0.1:4222:4222 nats:2.15.0-alpine -js
DDD_TS_KIT_TEST_NATS_URL=nats://127.0.0.1:4222 npx vitest run packages/nats
docker rm -f ddd-ts-kit-test-nats

CI的集成作业在到达这个包的每个拉取请求上都在真实服务器上运行它们， 针对NATS Server 2.15.0的JetStream pipeline/github/scripts/start-nats.sh 开始，它的形象被文摘钉住了。

他们在文件和内存存储上运行合约套件，然后是它的行为 提供者，每个行为一个套件(publishing，内外都有重复数据删除功能 重复的窗口， consuming， correlation， retries， dead-letters， topology， connection-loss 通过TCP中继测试切割， closing)和整个链条 发送给幂等消费者。每个测试都在它自己的流上运行 删除它们。