29 lines
578 B
Go
29 lines
578 B
Go
|
package initialize
|
||
|
|
|
||
|
|
import (
|
||
|
|
"context"
|
||
|
|
"github.com/twmb/franz-go/pkg/kgo"
|
||
|
|
"time"
|
||
|
|
)
|
||
|
|
|
||
|
|
func KafkaConsumerInit(ctx context.Context, config Config) (*kgo.Client, error) {
|
||
|
|
kafkaClient, err := kgo.NewClient(
|
||
|
|
kgo.SeedBrokers(config.KafkaBrokers),
|
||
|
|
kgo.ConsumerGroup("amoCRM"),
|
||
|
|
kgo.DefaultProduceTopic(config.KafkaTopic),
|
||
|
|
kgo.ConsumeTopics(config.KafkaTopic),
|
||
|
|
kgo.ConsumeResetOffset(kgo.NewOffset().AfterMilli(time.Now().UnixMilli())),
|
||
|
|
)
|
||
|
|
if err != nil {
|
||
|
|
return nil, err
|
||
|
|
}
|
||
|
|
|
||
|
|
err = kafkaClient.Ping(ctx)
|
||
|
|
if err != nil {
|
||
|
|
return nil, err
|
||
|
|
}
|
||
|
|
|
||
|
|
return kafkaClient, nil
|
||
|
|
|
||
|
|
}
|