const kafka = require("kafka-node");
const uuid = require("uuid");
 
const client = new kafka.Client("localhost:2181", "my-client-id", {
    sessionTimeout: 300,
    spinDelay: 100,
    retries: 2
});
 
client.on('ready', function (){
    console.log('producer client ready');
})  

client.on('error', function (err){
    console.log('client error: ' + err);
}) 
const producer = new kafka.HighLevelProducer(client);
producer.on("ready", function() {
    
    console.log("Kafka Producer is connected and ready.");
    /* KafkaService.sendRecord({  namespace: 'b',  room: 'c', channel: 'c', data: 'd' }, function (err, data) {
        console.log('incallback')
        console.log(err, data)
        producer.close();
        process.exit()
    }); */
});
 
// For this demo we just log producer errors to the console.
producer.on("error", function(error) {
    console.error(error);
});
 
const KafkaService = {
    sendRecord: ({  namespace, room, channel, data }, callback = () => {}) => {
        if (!namespace) {
            //return callback(new Error(`A namespace must be provided.`));
        }
 
        const event = {
            id: uuid.v4(),
            timestamp: Date.now(),
            namespace: namespace,
            room: room,
            channel: channel,
            data: data
        };
 
        const buffer = new Buffer.from(JSON.stringify(event));
 
        // Create a new payload
        const record = [
            {
                topic: "tcp.realtime",
                messages: buffer,
                attributes: 1 /* Use GZip compression for the payload */
            }
        ];
 
        //Send record to Kafka and log result/error
        producer.send(record, function (err, data) {
            //console.log(err, data);
        });
    }
};
 
module.exports = KafkaService;