Nestjs kafka implementation

Question

I've read nestjs microservice and kafka documentation but I couldn't figure out some of it. I'll be so thankful if you can help me out. So as the docs says I have to create a microService in main.ts file as follows:

const app = await NestFactory.createMicroservice<MicroserviceOptions>(AppModule, {
  transport: Transport.KAFKA,
  options: {
    client: {
      brokers: ['localhost:9092'],
    }
  }
});

await app.listen(() => console.log('app started'));

Then there is a kafkaModule file like this:

@Module({
  imports: [
    ClientsModule.register([
      {
        name: 'HERO_SERVICE',
        transport: Transport.KAFKA,
        options: {
          client: {
            clientId: 'hero',
            brokers: ['localhost:9092'],
          },
          consumer: {
            groupId: 'hero-consumer'
          }
        }
      },
    ]),
  ]
})
export class KafkaModule implements OnModuleInit {
  constructor(@Inject('HERO_SERVICE') private readonly clientService: KafkaClient)
  async onModuleInit() {
    await this.clientService.connect();
  }
}

The first thing I can't figure is what is the use of the first parameter of createMicroservice ? (I passed AppModule and KafkaModule and both worked correctly. knowing that kafkaModule is imported at appModule)

The other thing is that from what I understood, the microservice part and the configuration in the main.ts file is used to subscribe on the topics that is used in MessagePattern or EventPattern decorators, and the kafkaClient described in the kafkaModule is used to send messages to different topics.

the problem here is if what I said earlier is true, then why clientModule uses a default groupId if not specified to work as consumer. strange thing is I couldn't find a solution to get any message from any topic using clientModule. what I'm doing right now is to use different group ids in each file so they wont have any conflicts.

Luan Nguyen · Answer 1 · 2021-09-03T12:27:19.197

The first parameter of createMicroservice, it will help to guide how the consumer will connect to Kafka when you want to consume a message from a specific topic.

Example: we want to get message from topic: test01

How we declare?

import {Controller} from '@nestjs/common'
import {MessagePattern, Payload} from '@nestjs/microservices'

@Controller('sync')
export class SyncController {
  @MessagePattern('test01')
  handleTopicTest01(@Payload() message: Sync): any {
  // Handle your message here
  }
}

The second block is used as producer that is not consumer. When application want to send message to a specific topic, the clientModel will support this.

@Get()
sayHello() {
    return this.clientModule.send('say.hello', 'hello world')
}

Nestjs kafka implementation

1 Answers1