const { Kafka } = require('kafkajs');
const kafka = new Kafka({
clientId: 'sui-price-feed-consumer',
brokers: ['your-kafka-broker:9092'], // Replace with our Kafka broker address
});
const topic = 'sui-price-feed';
const consumer = kafka.consumer({ groupId: 'price-feed-group' });
const run = async () => {
await consumer.connect();
await consumer.subscribe({ topic, fromBeginning: false });
await consumer.run({
eachMessage: async ({ message }) => {
const payload = JSON.parse(message.value.toString());
console.log(`Received price update for ${payload.symbol}:`);
console.log(`Price: ${payload.price}`);
console.log(`Timestamp: ${new Date(payload.timestamp).toISOString()}`);
},
});
};
run().catch(console.error);