LibRdKafkaService.ts

LibRdKafkaService.ts

import { Injectable } from "@nestjs/common";
import Kafka from 'node-rdkafka';
import { ConfigService } from '@nestjs/config';

@Injectable()
export class LibRdKafkaService {
    private consumer: Kafka.KafkaConsumer;
    private producer: Kafka.Producer;

    constructor(private configService: ConfigService){
        
        this.consumer = new Kafka.KafkaConsumer({
            'metadata.broker.list': this.configService.get<string>('librdkafka-config.consumer.metadata-broker-list'),
            'group.id': this.configService.get<string>('librdkafka-config.consumer.group-id')

        }, {})

        this.producer = new Kafka.Producer({
            'metadata.broker.list': this.configService.get<string>('librdkafka-config.producer.metadata-broker-list'),
            'client.id': this.configService.get<string>('librdkafka-config.producer.client-id')            
        })

        this.consumer.connect()
        this.consumer
            .on('ready', () => {
                this.consumer.subscribe(['xxx']);
                this.consumer.consume();
            })
            .on('data', (data) => {
                console.log("[+] LibRdKafkaService", data.value.toString());
            });
    }

    getConsumer() : Kafka.KafkaConsumer 
    {
        return this.consumer;
    }

    getProducer() : Kafka.Producer 
    {
        return this.producer;
    }

}

// librdkafka-config:
//   producer:
//     client-id: "kafka"
//     metadata-broker-list: "localhost:9092"
//   consumer: 
//     group-id: "kafka",
//     metadata-broker-list: "localhost:9092"