Untitled

// Importação das bibliotecas necessárias
const { Kafka } = require('kafkajs');
const WebSocket = require('ws');

// Criação de uma instância do Kafka
const kafka = new Kafka({
 clientId: 'my-app', // Identificador único para esta aplicação
 brokers: ['localhost:9092'] // Endereço dos brokers do Kafka
});

// Criação de um consumidor Kafka
const consumer = kafka.consumer({ groupId: 'test-group' });

// Criação de um servidor WebSocket na porta 8080
const wss = new WebSocket.Server({ port: 8080 });

// Evento disparado quando um cliente se conecta ao servidor WebSocket
wss.on('connection', ws => {
 console.log('Client connected');

 // Inscrição no tópico do Kafka
 await consumer.subscribe({ topic: 'my-topic', fromBeginning: true });

 await consumer.run({
    eachMessage: async ({ topic, partition, message }) => {
      // Enviar a mensagem para o cliente conectado
      ws.send(message.value.toString());
    },
 });
});