forked from sodafoundation/strato
/
builder.go
55 lines (44 loc) · 1.27 KB
/
builder.go
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
package kafka
import (
"errors"
"github.com/confluentinc/confluent-kafka-go/kafka"
"github.com/opensds/multi-cloud/s3/pkg/helper"
bus "github.com/opensds/multi-cloud/s3/pkg/messagebus"
"github.com/opensds/multi-cloud/s3/pkg/messagebus/types"
)
type KafkaBuilder struct {
}
func (kb *KafkaBuilder) Create(config helper.MsgBusConfig) (bus.MessageSender, error) {
brokerList := ""
autoOffsetStore := false
params := config.Server
if nil == params || len(params) == 0 {
return nil, errors.New("input kafka params is invalid")
}
// get broker list
v, ok := params[types.KAFKA_CFG_BROKER_LIST]
if !ok {
return nil, errors.New("params doesn't contain validate broker list")
}
brokerList = v.(string)
// try to get the enable.auto.offset.store
if v, ok = params[types.KAFKA_CFG_AUTO_OFFSET_STORE]; ok {
autoOffsetStore = v.(bool)
}
p, err := kafka.NewProducer(&kafka.ConfigMap{
"bootstrap.servers": brokerList,
"enable.auto.offset.store": autoOffsetStore,
"request.timeout.ms": config.RequestTimeoutMs,
"message.timeout.ms": config.MessageTimeoutMs,
"message.send.max.retries": config.SendMaxRetries,
})
if err != nil {
return nil, err
}
kafka := &Kafka{
producer: p,
doneChan: make(chan int),
}
kafka.Start()
return kafka, nil
}