-
Notifications
You must be signed in to change notification settings - Fork 4
/
kafka_provider.go
59 lines (51 loc) · 1.28 KB
/
kafka_provider.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
56
57
58
59
package inputstreams
import (
"github.com/arquivei/foundationkit/errors"
"github.com/arquivei/goduck"
"github.com/arquivei/goduck/impl/implstream/kafkaconfluent"
"github.com/confluentinc/confluent-kafka-go/v2/kafka"
)
type kafkaProvider struct {
topic []string
configMap kafka.ConfigMap
}
// WithKafkaProvider configures the input stream with a kafka provider.
func WithKafkaProvider(opts ...KafkaOption) Option {
return func(o *options) error {
// Create kafka provider
provider := &kafkaProvider{
configMap: kafka.ConfigMap{
"auto.offset.reset": "earliest",
"enable.auto.offset.store": "false",
},
}
for _, opt := range opts {
opt(provider)
}
if len(provider.topic) == 0 {
return ErrNoKafkaTopic
}
for _, t := range provider.topic {
if t == "" {
return ErrEmptyKafkaTopic
}
}
o.provider = provider
return nil
}
}
func (p *kafkaProvider) MakeStream() (s goduck.Stream, err error) {
// KLUDGE: goduck doesn't provide a constructor that doesn't panic.
defer func() {
if r := recover(); r != nil {
err = errors.NewFromRecover(r)
}
}()
s = kafkaconfluent.MustNew( // FIXME: Add new on goduck and refactor here
kafkaconfluent.Config{
Topics: p.topic,
RDKafkaConfig: &p.configMap,
},
)
return s, nil
}