如何设置NestJS微服务集成/e2e测试KafkaJS消费者（MessagePattern/EventPattern处理程序）

Question

我正在验证与 NestJS/KafkaJS 的集成测试。

除了我要发送到的主题的事件监听器（消费者）上的函数没有被调用之外，我已经实现了所有内容。

我在某处读到，在消费者事件 GROUP_JOIN 完成之前你无法使用消息，不确定这是否正确，和/或我如何让我的 e2e 测试等待这种情况发生？

这是 e2e 测试的设置 -

describe('InventoryController(e2e)', () => {
  let app: INestApplication
  let client: ClientKafka
  beforeAll(async () => {
    const moduleFixture: TestingModule = await Test.createTestingModule({
      imports: [InventoryModule, KafkaClientModule],
    }).compile()
    app = moduleFixture.createNestApplication()
    await app.connectMicroservice({
      transport: Transport.KAFKA,
      options: {
        client: {
          clientId: 'test_clientid',
          brokers: process.env.KAFKA_BROKERS.split(' '), // []
          ssl: true,
          sasl: {
            mechanism: 'plain',
            username: process.env.KAFKA_CLUSTER_APIKEY,
            password: process.env.KAFKA_CLUSTER_SECRET,
          },
        },
        consumer: {
          groupId: 'test_consumerids',
        },
      },
    })
    await app.startAllMicroservices()
    await app.init()
    client = moduleFixture.get<ClientKafka>('test_name')
    await client.connect()
    await app.listen(process.env.port || 3000)
  })

  afterAll(async () => {
    await app.close()
    await client.close()
  })

  it('/ (GET)', async () => {
    return request(app.getHttpServer()).get('/inventory/kafka-inventory-test')
  })

  it('Emits a message to a topic', async () => {
    await client.emit('inventory-test', { foo: 'bar' })
  })

客户端正在正常发出消息，

在我的控制器中，我有主题“库存测试”的事件处理程序

  @EventPattern('inventory-test')
  async consumeInventoryTest(
    // eslint-disable-next-line
    @Payload() inventoryMessage: any,
    @Ctx() context: KafkaContext,
  ): Promise<void> {
    console.log('inventory-test consumer')
  }

我还使用 app.getMicroservices() 方法记录了微服务，并且可以在 messageHandlers 对象下看到它有“inventory-test”，它返回一个函数

        server: ServerKafka {
          messageHandlers: Map(1) { 'inventory-test' => [Function] }

当我在本地运行应用程序时，消息处理程序正在工作。

我在 Google 上搜索了很多 kafkajs 和 Nest 的文档，但没有太多信息

Answer 1

我实际上终于解决了这个问题，在客户端发出消息后，您需要等待新的承诺，以便处理程序有时间读取它。