package service

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

var (
	//agentBrokers    = []string{"192.168.4.100:15091", "192.168.4.100:15092", "192.168.4.100:15093"}
	agentBrokers    = []string{"172.30.12.19:9092", "172.30.12.20:9092", "172.30.12.21:9092"}
	agentKafkaTopic = "quantGroup.tech.enoch.pro"
	agentKafkaGroup = "quantGroup-enoch-agent"
)

type enochAgentConsumerGroupHandler struct{}

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

func AgentConsumer() {
	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(agentBrokers, config)

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

	consumerGroup, err := sarama.NewConsumerGroupFromClient(agentKafkaGroup, 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 := enochAgentConsumerGroupHandler{}
	for {
		err = consumerGroup.Consume(ctx, []string{agentKafkaTopic}, handler)

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

}
