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
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
|
package main
import (
"context"
"fmt"
"log"
"os"
"github.com/mongodb/mongo-go-driver/mongo"
kafka "github.com/segmentio/kafka-go"
)
func getMongoCollection(mongoURL, dbName, collectionName string) *mongo.Collection {
client, err := mongo.Connect(context.Background(), mongoURL)
if err != nil {
log.Fatal(err)
}
// Check the connection
err = client.Ping(context.Background(), nil)
if err != nil {
log.Fatal(err)
}
fmt.Println("Connected to MongoDB ... !!")
db := client.Database(dbName)
collection := db.Collection(collectionName)
return collection
}
func getKafkaReader(kafkaURL, topic, groupID string) *kafka.Reader {
return kafka.NewReader(kafka.ReaderConfig{
Brokers: []string{kafkaURL},
GroupID: groupID,
Topic: topic,
MinBytes: 10e3, // 10KB
MaxBytes: 10e6, // 10MB
})
}
func main() {
// get Mongo db Collection using environment variables.
mongoURL := os.Getenv("mongoURL")
dbName := os.Getenv("dbName")
collectionName := os.Getenv("collectionName")
collection := getMongoCollection(mongoURL, dbName, collectionName)
// get kafka reader using environment variables.
kafkaURL := os.Getenv("kafkaURL")
topic := os.Getenv("topic")
groupID := os.Getenv("groupID")
reader := getKafkaReader(kafkaURL, topic, groupID)
defer reader.Close()
fmt.Println("start consuming ... !!")
for {
msg, err := reader.ReadMessage(context.Background())
if err != nil {
log.Fatal(err)
}
insertResult, err := collection.InsertOne(context.Background(), msg)
if err != nil {
log.Fatal(err)
}
fmt.Println("Inserted a single document: ", insertResult.InsertedID)
}
}
|
