@ eco- foundation/ eco- message

用于生态服务的硬巢JS AMQP运输:连接寿命周期、地形作为代码、a 基于确认的出版商,带有中继功能的交易出柜机,以及同能消费者主机. 提取后, 正确的发布/ 假设行为是默认的, 而不是每个服务的东西 自行再审再审.

pnpm add @eco-foundation/eco-messaging @eco-foundation/eco-message-contracts

@eco-foundation/eco-message-contracts 是一个 端点,而不是依赖,所以单独安装这个软件包会留下一个未满足的对等点——同时安装两个 当然 剩下的同伴是雀巢JS服务已经拥有的东西, 并且是同伴 所以这个库 从未将您的框架或 AMQP 客户端的第二个副本拖入树上 : @nestjs/common (单位:千美元)^10 || ^11, (中文(简体) ), @nestjs/core (单位:千美元)^10.2 || ^11, (中文(简体) ), amqplib (单位:千美元)^0.10, (中文(简体) ), amqp-connection-manager (单位:千美元)^5, (中文(简体) ), 财务报告和审定财务报表 zod (单位:千美元)^3.25 || ^4) (中文(简体) ). @opentelemetry/api 财务报告和审定财务报表 testcontainers 是可选的同伴,仅需要 或导入 /testing 子路径.

说明 zod 楼层 : ^3.25.0. 申报较大范围的服务需要提高申报范围 即使它的锁文件已经解决了新的问题.

担保

最少一次交货加一能消费. 这个图书馆不提供准确的一次 交付——这种原始的在这里并不存在——所以你的操作者必须安全地运行两次. 这是什么 库中提供的信息是,在您的处理器解决之前,信件不被确认,即 重复发送同一信件的ID被跳过而不是重新应用,失败 重复通过中介地貌学而不是在过程中入睡.

把它连接起来

import { Module } from '@nestjs/common';
import {
  EcoMessagingModule,
  MongoOutbox,
  MongoIdempotencyLedger,
} from '@eco-foundation/eco-messaging';

@Module({
  imports: [
    EcoMessagingModule.forRoot({
      url: process.env.RABBITMQ_URL!,
      exchange: 'my-service.events',
      sourceService: 'my-service',
      // Host-supplied, already-connected. The library never opens its own datastore
      // connection — it piggybacks on yours and owns only its own collections.
      outbox: MongoOutbox.forConnection(mongooseConnection),
      ledger: MongoIdempotencyLedger.forConnection(mongooseConnection),
    }),
  ],
})
export class AppModule {}

forRoot编号 :forRootAsync (无) register),目的:每架有1个通讯连接 服务,方法名称也这么说。 模块是 @Global ——在应用根上导入一次. 每个功能模块导入它,是服务最终如何有两个连接,心跳是 每一个连接,所以一个出版商可以看起来健康,而另一个消费者则已经死亡.

只有出版商的服务可以省略 outbox编号 :ledger 完全; 内存出入口已接通 自动。 向消费者登记 没有 分类账是一个被拒绝的配置——消耗 不允许无声无息地退化.

持久性是一个带有适配器的端口,主机选择: MemoryOutbox编号 :MongoOutbox 财务报告和审定财务报表 MemoryLedger编号 :MongoIdempotencyLedger编号 :RedisIdempotencyLedger。图书馆从不打开自己的 数据存储连接, 永不调用 mongoose.connect(),没有 mongoose 相对依赖关系——什么 由Mongoose各大专业的服务分享.

出版

制片人默认通过发件箱出版,并用于 required 合同不仅仅是 建议——这是唯一被认可的路径:

import { obligationFromContract } from '@eco-foundation/eco-messaging';

await outbox.record(
  obligationFromContract(InventorySnapshot, payload, {
    now: new Date(),
    sourceService: 'my-service',
  }),
  dbSession, // your own transaction handle, so the record commits atomically with the event
);

publisher.publishNowOrThrow(contract, payload) 是已关闭的直接路径 : 证实并扔在 nack,返回,超时, 或断开,而不是排队任何东西。 它承认 required 只有合同宣布 durability: 'caller-ledger'。 。 。 .

如果失败, EcoPublishError 答题二 直肠 问. isRetryablePublicationFailure 问“能否重试此信息是否成功?”; wireOutcome 问经纪人有没有见过 ”——这就是决定失败是否应该花费有限的尝试预算的原因。 打电话的人 将三线结果减为两线必须切 not-attempted 永远不要 broker-verdict 对万事俱备;后者允许含糊不清地发表无拘束的重试.

如果您在发布前租入了债务,则租入的TTL的大小 maxPublishWallTimeMs(options)过去的束缚 缓慢但活的出版社的租约被偷了 中途飞行和租赁开始 生产 复数.

消费

import { Injectable } from '@nestjs/common';
import type { HandlerContext } from '@eco-foundation/eco-messaging';
import {
  EcoConsumer,
  EcoMessageHandlerProvider,
  TransientError,
} from '@eco-foundation/eco-messaging';

@Injectable()
@EcoConsumer({ queue: 'my-service.snapshots', contracts: [InventorySnapshot] })
export class SnapshotHandler implements EcoMessageHandlerProvider {
  async handle(payload: unknown, context: HandlerContext): Promise<void> {
    // Your resolve is your durability attestation: the ack happens only after this resolves.
    // Throw TransientError to retry via topology-based backoff, or PermanentError to park
    // immediately in the DLQ.
    if (somethingRecoverable) throw new TransientError('downstream unavailable, retry me');
  }
}

消费者注册实现a 三进制 每次重联时的队列: X.v2 (单位:千美元) 法定工作排队, X.v2.retry (TTL,死字写回工作),和 X.v2.dlq (停车场, 别再写了 在没有三重奏的情况下签字是一种被拒绝的配置.

对于一个生产者拥有的,未开发的,最新价值的饲料,你不控制, EcoExternalSnapshotConsumer 在共享的连接上读取它——一个有界限的 x-max-length: 1 队列, 失败会重排一次 然后下降,因为被重审的消息是有保障的 stale.

准备情况

线 EcoMessagingHealth.isReady() 进入你的探测器,不是 EcoConnectionManager.isConnected()(单位:千美元) 队列不存在的连接客户端尚未准备好消费, 并且报告状态已准备就绪 如何静静地部署黑洞交通.

准备状态是 命名大门。该图书馆拥有 topology; 重建服务 靴子上的记忆状态会自行登记,所以"运行"和"补液"仍然可以区分:

this.rehydration = health.registerGate('rehydration'); // registered closed
// ...later
this.rehydration.open(`rebuilt ${restored} obligations`);

一个封闭的大门扣住 只准备 永远不要消费, 也不要排出器的排出. 两者都是故意的:一个服务,其还水来源是公共汽车,否则会陷入僵局,而且 无论重建需要多长时间,在崩溃前承担的义务必须公布.

EcoMessagingStatus 是主机连接和发布器状态的单独事件视图 生命周期适配器(瞭望犬,准备拉链,工人门). 它什么也没有, 证书是 在任何细节到达听众之前已编辑.

观察性

ALERT 输出五个结构化的对数行文字。 生态警报 基于日志 - 这个图书馆的 自定义 OTLP 度量衡不能到达 Datadog —— 所以这五个字符串是唯一可以按到 an 正在呼叫的工程师 有关一个消息故障。 它们被测试用字节表示; 重命名一个是一个 打破更改.

METRIC 财务报告和审定财务报表 METRIC_KINDS 指定仪器名称,涵盖耐久性的一半: (单位:千美元)publish.*, (中文(简体) ), consume.*, (中文(简体) ), outbox.*)和呼叫者引导者承担的生命周期 (单位:千美元)obligation.state, (中文(简体) ), obligation.pending, (中文(简体) ), obligation.oldest_age_ms) (中文(简体) ). 无测量的测量仪 报告只有零.

子路径导出

如果给OTel接线,请读取命令中的两个陷阱 仓库读取器 第一——登记 MeterProvider 在建立测量槽之后,无声无息地产生任何测量值.

从1.x升级

2.0.0是地形学和地表学方面的一个主要原因;目前没有什么改变行为和新的界限 默认为主机已经拥有 。 有一个断裂源变化: RawPublishInput.envelope 实值 现在 Omit<Envelope, 'correlationId'>,因为 publishRawOrThrow 总是重新铸造 每一次尝试的关联身份 并丢弃任何通过。 一个呼叫者通过一个完整的 Envelope 对象字元必须去掉字段。 见 变动情况 联 合 国 完全进入.

文档

这些包是标准的参考执行,而不是其定义. 定义是 两种语言不可知文件,两种机器都根据这个代码检查:

• 信使符合配置 ——信封,地貌,结算,解除,订购,可观察.
• 推前坚持和还水 ——生产耐用性和就绪性.
• 警报 - 每一个 ALERT 及其监控状态.
• 仓库读取器 完整指南,包括 .v2 典型的-q 法定人数削减规则、蓄意超出范围的事项以及公认的剩余 在接触连接或消费者生命周期前值得阅读的风险.