雀巢JS SQS

测试过 AWS 质量, (中文(简体) ), 弹性MQ,以及 本地软件包 (参见本repo中的整合测试).

NestJS SQS是一个使SQS与NestJS一起更容易使用和控制一些需要的流量的项目. 本单元提供 基于装饰器的信件处理适合简单使用.

这个图书馆内部依赖 bbc/sqs-生产者 财务报告和审定财务报表 bbc/sqs - 消费者,并在 他们给予的基本功能.

默认 SqsSerializer 财务报告和审定财务报表 SqsDeserializer 用于标准队列的工作。 消息 机构 对 FIFO 是相同的 JSON; FIFO 元数据在 SQS 消息上, 不在 Body,所以只有 序列器 fIFO 的更改: 注册 SqsFifoSerializer 通过 serializer 选项 SqsClient 编号 : SqsServer 需要时(全部从此包导出).

比起Redis和RabbitMQ运输机

巢穴的鱼 重写 财务报告和审定财务报表 季度 服务器是第一当事方:多路连接、推式交付和经纪人记录上丰富的元数据. 质量 设计不同:

巢穴 API 密码 保持熟悉: @MessagePattern 编号 : @EventPattern, (中文(简体) ), ClientProxy和序列器——但是 操作行为 遵循 SQS(可见超时, DLQ, 减速). 计划相应重审并具"一能".

所需资源

• 节点.js 24+ (将运行于22日,但不测试)

安装

$ npm i --save @ethberry/nestjs-sqs

当地融合测试

已读集成规格 ELASTICMQ_URL 财务报告和审定财务报表 LOCALSTACK_URL 从a .env 文件位于 repo 根(承诺默认值)。 启动弹性MQ和本地Stack,然后进行测试:

$ npm i
$ npm run docker:sqs:up
$ npm test
$ npm run docker:sqs:down

通过编辑覆盖终点 .env (例如不同的端口).

要匹配CI( lint + 测试) 在依赖度和多克值上升之后:

$ npm run lint && npm test

快速启动

海关运输登记

就像你登记其他微服务一样

import { SQSClient } from "@aws-sdk/client-sqs";
import { SqsServer } from "@ethberry/nestjs-sqs";

const sqs = new SQSClient({
  endpoint: "http://localhost:9324",
  region: "none",
  credentials: {
    accessKeyId: "x",
    secretAccessKey: "x",
  },
});

app.connectMicroservice({
  strategy: new SqsServer({
    consumerOptions: {
      sqs,
      queueUrl: "http://localhost:9324/queue/producer",
    },
    producerOptions: {
      sqs,
      queueUrl: "http://localhost:9324/queue/consumer",
    },
  }),
});

装饰方法

您需要在您的 NestJS 控制器中解析方法, 以便它们作为事件处理器自动附加 用于接收 SQS 信件 :

@Controller()
export class SqsController {
  @MessagePattern(MESSAGE_TYPE)
  public handleMessage(message: any): Promise<any> {
    // do something, return result
  }

  @EventPattern(EVENT_TYPE)
  public handleEvent(event: any): Promise<void> {
    // do something
  }
}

处理器上下文SqsContext(中文(简体) )

喜欢 RmqContext 或者说 RedisContext时,您可以注射 SqsContext 与 @Ctx() 读取原始 AWS Message消费者 队列 URL,则 模式 绳子。 第一个论点仍然是被破坏 有效载数 (单位:千美元)data 从身体上).

import { Controller } from "@nestjs/common";
import { Ctx, MessagePattern, Payload } from "@nestjs/microservices";
import { SqsContext } from "@ethberry/nestjs-sqs";

@Controller()
export class SqsController {
  @MessagePattern(MESSAGE_TYPE)
  public handleRpc(@Payload() data: unknown, @Ctx() ctx: SqsContext): Promise<unknown> {
    const sqs = ctx.getMessage();
    const attrs = sqs.MessageAttributes;
    // ...
    return { ok: true };
  }
}

无顶层排放的事件 id 在数据包中以 @EventPattern (想法和其他巢穴运输一样).

伟大的关闭

sqs-consumer 可以等待最后的民调和飞行 handleMessage 停止前工作。 设定 pollingCompleteWaitTimeMs 打开 consumerOptions (毫秒,你愿意等待) . SqsServer.close() 转发可选 StopOptions 改为 consumer.stop() (例如 { abort: true } (b) 取消飞行中的民意调查——节制使用).

new SqsServer({
  consumerOptions: {
    sqs,
    queueUrl: "...",
    pollingCompleteWaitTimeMs: 10_000,
  },
  producerOptions: { sqs, queueUrl: "..." },
});

当应用停止,打电话 app.close() (或以其他方式援引微服务) close)所以消费者关闭.

产生信件

import { SQSClient } from "@aws-sdk/client-sqs";
import { SqsClient, SQS_SERVICE } from "@ethberry/nestjs-sqs";

const sqs = new SQSClient({
  endpoint: "http://localhost:9324",
  region: "none",
  credentials: {
    accessKeyId: "x",
    secretAccessKey: "x",
  },
});

@Module({
  imports: [
    ClientsModule.register([
      {
        name: SQS_SERVICE,
        customClass: SqsClient,
        options: {
          consumerUrl: "http://localhost:9324/queue/consumer",
          producerUrl: "http://localhost:9324/queue/producer",
          sqs,
        },
      },
    ]),
  ],
  providers: [AppService],
})
class AppModule {}

export class AppService {
  constructor(
    @Inject(SQS_SERVICE)
    private readonly sqsClientProxy: ClientProxy,
  ) {}

  public dispatch(): Promise<void> {
    void this.sqsClientProxy.emit(EVENT_NAME, {});
  }
}

许可证

这个项目根据麻省理工学院许可证的规定获得许可.