package service

import (
	"context"
	"fmt"
	"github.com/Shopify/sarama"
	"time"
)

var (
	brokers    = []string{"172.30.12.19:9092", "172.30.12.20:9092", "172.30.12.21:9092"}
	kafkaTopic = "quantGroup.tech.brave.pro"
	kafkaGroup = "another-group-test"
)

type exampleConsumerGroupHandler struct{}

func (exampleConsumerGroupHandler) Setup(_ sarama.ConsumerGroupSession) error   { return nil }
func (exampleConsumerGroupHandler) Cleanup(_ sarama.ConsumerGroupSession) error { return nil }
func (h exampleConsumerGroupHandler) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
	for msg := range claim.Messages() {
		MsgProcess(string(msg.Value))
		sess.MarkMessage(msg, "")
	}
	return nil
}

func Consumer() {
	config := sarama.NewConfig()

	config.Version = sarama.V1_1_0_0
	config.Consumer.Retry.Backoff = 100 * time.Second
	config.Consumer.Return.Errors = false

	client, err := sarama.NewClient(brokers, config)

	if err != nil {
		fmt.Printf("kafka 消费者获取失败")
		return
	}
	defer func() { _ = client.Close() }()

	consumerGroup, err := sarama.NewConsumerGroupFromClient(kafkaGroup, client)
	if err != nil {
		fmt.Println("kafka 创建分区消费者失败")
		return
	}
	defer func() { _ = consumerGroup.Close() }()

	// Track errors
	go func() {
		for err := range consumerGroup.Errors() {
			fmt.Println("消费出错了", err)
		}
	}()
	ctx := context.Background()
	handler := exampleConsumerGroupHandler{}

	for {
		err = consumerGroup.Consume(ctx, []string{kafkaTopic}, handler)

		if err != nil {
			fmt.Println(err)
		}
	}

}
