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
73
74
75
76
77
78
79
80
81
82
83
|
package replication
import (
"fmt"
"github.com/Shopify/sarama"
"github.com/chrislusf/seaweedfs/weed/glog"
"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
"github.com/chrislusf/seaweedfs/weed/util"
"github.com/golang/protobuf/proto"
)
func init() {
NotificationInputs = append(NotificationInputs, &KafkaInput{})
}
type KafkaInput struct {
topic string
consumer sarama.Consumer
messageChan chan *sarama.ConsumerMessage
}
func (k *KafkaInput) GetName() string {
return "kafka"
}
func (k *KafkaInput) Initialize(configuration util.Configuration) error {
glog.V(0).Infof("replication.notification.kafka.hosts: %v\n", configuration.GetStringSlice("hosts"))
glog.V(0).Infof("replication.notification.kafka.topic: %v\n", configuration.GetString("topic"))
return k.initialize(
configuration.GetStringSlice("hosts"),
configuration.GetString("topic"),
)
}
func (k *KafkaInput) initialize(hosts []string, topic string) (err error) {
config := sarama.NewConfig()
config.Consumer.Return.Errors = true
k.consumer, err = sarama.NewConsumer(hosts, config)
if err != nil {
panic(err)
} else {
glog.V(0).Infof("connected to %v", hosts)
}
k.topic = topic
k.messageChan = make(chan *sarama.ConsumerMessage, 1)
partitions, err := k.consumer.Partitions(topic)
if err != nil {
panic(err)
}
for _, partition := range partitions {
partitionConsumer, err := k.consumer.ConsumePartition(topic, partition, sarama.OffsetNewest)
if err != nil {
panic(err)
}
go func() {
for {
select {
case err := <-partitionConsumer.Errors():
fmt.Println(err)
case msg := <-partitionConsumer.Messages():
k.messageChan <- msg
}
}
}()
}
return nil
}
func (k *KafkaInput) ReceiveMessage() (key string, message *filer_pb.EventNotification, err error) {
msg := <-k.messageChan
key = string(msg.Key)
message = &filer_pb.EventNotification{}
err = proto.Unmarshal(msg.Value, message)
return
}
|
