-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathconsumer.js
38 lines (31 loc) · 924 Bytes
/
consumer.js
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
const {Kafka} = require("kafkajs");
run();
async function run() {
try {
// Create an admin connection
const kafka = new Kafka({
"clientId": "myapp",
"brokers": ["pmventura:9092"]
});
const consumer = kafka.consumer({groupId: "test"});
console.log("Connnectingg......");
await consumer.connect();
console.log("Connected!");
// Read all the time
await consumer.subscribe({
"topic": "Users",
"fromBeginning": true
});
await consumer.run({
// This will gonna process each message has been received
"eachMessage": async result => {
console.log(`RVD Msg ${result.message.value} on partition ${result.partition}`)
}
});
}
catch(ex) {
console.error(`Something bad happened ${ex}`)
}
finally {
}
}