Function startServer

integration/microservices/e2e/concurrent-kafka.spec.ts:25–62

()

  this.timeout(30000);

  const startServer = async () => {
    const module = await Test.createTestingModule({
      controllers: [
        KafkaConcurrentController,
        KafkaConcurrentMessagesController,
      ],
    }).compile();

    // use our own logger for a little
    // Logger.overrideLogger(new Logger());

    const app = module.createNestApplication();

    const server = app.getHttpAdapter().getInstance();

    app.connectMicroservice<MicroserviceOptions>({
      transport: Transport.KAFKA,
      options: {
        client: {
          brokers: ['localhost:9092'],
        },
        run: {
          partitionsConsumedConcurrently: numbersOfServers,
        },
      },
    });

    // enable these for clean shutdown
    app.enableShutdownHooks();

    // push to the collection
    servers.push(server);
    apps.push(app);

    // await the start
    await app.startAllMicroservices();
    await app.init();
  };

  it(`Create kafka topics/partitions`, async () => {
    const kafka = new Kafka({