package service

import (
	"context"
	"fmt"
	"github.com/Shopify/sarama"
	"github.com/bsm/sarama-cluster"
	"log"
	"os"
	"os/signal"
	"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 = "enoch-group"
)

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 ClusterConsumer() {
	config := cluster.NewConfig()

	config.Consumer.Return.Errors = true
	config.Consumer.Offsets.Initial = sarama.OffsetOldest
	config.Group.Return.Notifications = true

	consumer, err := cluster.NewConsumer(brokers, kafkaGroup, []string{kafkaTopic}, config)
	if err != nil {
		panic(err)
	}
	defer func() { _ = consumer.Close() }()
	// trap SIGINT to trigger a shutdown.
	signals := make(chan os.Signal, 1)
	signal.Notify(signals, os.Interrupt)

	// consume errors
	go func() {
		for err := range consumer.Errors() {
			log.Printf("Error: %s\n", err.Error())
		}
	}()

	// consume notifications
	go func() {
		for ntf := range consumer.Notifications() {
			log.Printf("Rebalanced: %+v\n", ntf)
		}
	}()

	// consume messages, watch signals
	for {
		select {
		case msg, ok := <-consumer.Messages():
			if ok {
				//fmt.Fprintf(os.Stdout, "%s/%d/%d\t%s\t%s\n", msg.Topic, msg.Partition, msg.Offset, msg.Key, msg.Value)
				MsgProcess(string(msg.Value))
				consumer.MarkOffset(msg, "") // mark message as processed
			}
		case <-signals:
			return
		}
	}

}

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)
		}
	}

}
