import { Injectable } from '@nestjs/common';

const { Kafka } = require('kafkajs')// kafka js npm pacakge

const kafka = new Kafka({
     
    brokers: [       
        'localhost:9093'// kafka service port
    ],
    // ssl: true,// if SSL is turned on
    // sasl: {
    //     mechanism: 'SCRAM-SHA-512', // scram-sha-256 or scram-sha-512
    //     username: 'username',
    //     password: 'password'
    // },// if auth is enabled the credentials will be set at this place
});
const producer = kafka.producer();
const consumer = kafka.consumer(
    { groupId: 'kafka-nodejs' }// define your own group id
);
@Injectable()
export class KafkaConnectorService {
  constructor(){
      this.init();  }

/**
 * 
 * init function wich will connect, send a initial message to the topic
 * and initialize the consumer listner
 * */
  async init(){
      await this.connect();
      await this.sendMessage();
      await this.listenToMessages();
  }

/**
 * connect to service
 * */
    async  connect(){
      console.log("connect.....")
        await consumer.connect();//conenct consumer/listneer
        await producer.connect();//conenct producer/emitter
        await consumer.subscribe({ topic: 'heapspace-out-topic', fromBeginning: true });         
        return  true;
    }

/**
 * send sample message to the topic
 * */
    async sendMessage(){
        console.log("sendMessage.....")
        await producer.send({
            topic: 'heapspace-out-topic',
            messages: [
                { value: JSON.stringify({
                        "name": "Sam",
                        "age": "34"
                         
                    }) },
            ],
        })
    }

/**
 * listen to any messages comming on the topic
 * */
    async listenToMessages(){
        console.log("listenToMessages.....")
        await consumer.run({
            eachMessage: async ({ topic, partition, message }) => {
                console.log('new msg');
                console.log(topic);
                console.log({
                    value: message.value.toString(),
                });
                console.log(JSON.parse(message.value))
                //once you get a message from the topic, you can do your own logic 
                //using the message recived
            },
        })
    }
}