NestJS's Dynamic Module for Consumer.
Install
npm install --save @kafka-ts/nestjs-consumer
# or
yarn add @kafka-ts/nestjs-consumer
# or
pnpm add @kafka-ts/nestjs-consumerUsage
// main.ts
import { NestFactory } from '@nestjs/core';
import { KafkaConsumer } from '@kafka-ts/nestjs-consumer';
import { AppModule } from 'app.module';
async function bootstrap() {
const app = await NestFactory.createMicroservice<MicroserviceOptions>(
AppModule,
{
strategy: new KafkaConsumer({
brokers: ['localhost:9092'],
consumerOptions: {
groupId: 'test-id',
},
}),
},
);
await app.listen(3_000);
}
bootstrap();// main.ts
import { NestFactory } from '@nestjs/core';
import { KafkaConsumer } from '@kafka-ts/nestjs-consumer';
import { AppModule } from 'app.module';
async function bootstrap() {
const app = await NestFactory.createMicroservice<MicroserviceOptions>(
AppModule,
KafkaConsumer.createService({
brokers: ['localhost:9092'],
consumerOptions: {
groupId: 'test-id',
},
}),
);
await app.listen(3_000);
}
bootstrap();// main.ts
import { NestFactory } from '@nestjs/core';
import { KafkaConsumer } from '@kafka-ts/nestjs-consumer';
import { AppModule } from 'app.module';
async function bootstrap() {
const app = await NestFactory.create(AppModule);
app.connectMicroservice(
KafkaConsumer.createService({
brokers: ['localhost:9092'],
consumerOptions: {
groupId: 'test-id',
},
}),
);
await app.startAllMicroservices();
await app.listen(3_000);
}
bootstrap();// consumer.controller.ts
import { Controller } from '@nestjs/common';
import {
Ctx,
Payload,
Subscribe,
SubscribeMessage,
KafkaMessageContext,
KafkaBatchMessageContext,
} from '@kafka-ts/nestjs-consumer';
@Controller()
export class ConsumerController {
@Subscribe({
topics: ['topic'],
})
public async handleSubscribe(
@Payload() data: string[],
@Ctx() context: KafkaBatchMessageContext,
): Promise<string> {
console.log('data', data);
console.log('context.batch', context.batch);
return 'Ok!';
}
}In case you wanna support multiple clientId
// main.ts
import { NestFactory } from '@nestjs/core';
import { KafkaConsumer } from '@kafka-ts/nestjs-consumer';
import { AppModule } from 'app.module';
async function bootstrap() {
const app = await NestFactory.create(AppModule);
app.connectMicroservice(
KafkaConsumer.createService([
{
brokers: ['localhost:9092'],
consumerOptions: {
groupId: 'test-id',
},
},
{
clientId: 'test-client',
brokers: ['localhost:9092'],
consumerOptions: {
groupId: 'test-id-2',
},
},
]),
);
await app.startAllMicroservices();
await app.listen(3_000);
}
bootstrap();// consumer.controller.ts
import { Controller } from '@nestjs/common';
import {
Ctx,
Payload,
Subscribe,
SubscribeMessage,
KafkaMessageContext,
KafkaBatchMessageContext,
} from '@kafka-ts/nestjs-consumer';
@Controller()
export class ConsumerController {
@Subscribe({
topics: ['topic'],
})
public async handleSubscribe(
@Payload() data: string[],
@Ctx() context: KafkaBatchMessageContext,
): Promise<string> {
console.log('data', data);
console.log('context.batch', context.batch);
return 'Ok!';
}
@Subscribe({
clientId: 'test-client',
topics: ['topic_2'],
})
public async handleSubscribeTestClient(
@Payload() data: string[],
@Ctx() context: KafkaBatchMessageContext,
): Promise<string> {
console.log('data', data);
console.log('context.batch', context.batch);
return 'Ok!';
}
}Wanna subscribe eachBatch and eachMessage?
If you wanna subscribe eachBatch and eachMessage, you should define two topic in different clientId or else only eachBatch or eachMessage can run.
// main.ts
import detect from 'detect-port';
import { NestFactory } from '@nestjs/core';
import { KafkaConsumer } from '@kafka-ts/nestjs-consumer';
import { AppModule } from './app.module';
async function bootstrap() {
const app = await NestFactory.create(AppModule);
app.connectMicroservice(
KafkaConsumer.createService([
{
brokers: ['localhost:9092'],
consumerOptions: {
groupId: 'test-id',
},
},
{
clientId: 'test-client',
brokers: ['localhost:9092'],
consumerOptions: {
groupId: 'test-id-2',
},
},
]),
);
const port = await detect(3_000);
await app.startAllMicroservices();
await app.listen(port);
console.log(`Run on ${port}`);
}
bootstrap();// consumer.controller.ts
import { Controller } from '@nestjs/common';
import {
Ctx,
Payload,
Subscribe,
SubscribeMessage,
KafkaMessageContext,
KafkaBatchMessageContext,
} from '@kafka-ts/nestjs-consumer';
@Controller()
export class ConsumerController {
@Subscribe({
topics: ['topic'],
})
public async handleSubscribe(
@Payload() data: string[],
@Ctx() context: KafkaBatchMessageContext,
): Promise<string> {
console.log('data', data);
console.log('context.batch', context.batch);
return 'Ok!';
}
@SubscribeMessage({
clientId: 'test-client',
topics: ['topic_2'],
})
public async handleSubscribeMessage(
@Payload() data: string,
@Ctx() context: KafkaMessageContext,
): Promise<string> {
console.log('data', data);
console.log('context.message', context.message);
return 'Ok!';
}
}