package kafka

import (
	"context"
	"fmt"
	conf "gitee.com/wuzheng0709/backend-gopkg/infrastructure/config"
	"github.com/Shopify/sarama"
	"strings"
)

const GolLogTopic = "global_log"

var (
	KafkaClient  sarama.ConsumerGroup
	KafkaProduct sarama.SyncProducer
)

func init() {
	var err error
	//KafkaClient, err = GetKafkaClient()
	//if err != nil {
	//	log.Error("生成 kafka 消费组实例失败，err:", err.Error())
	//	return
	//}
	KafkaProduct, err = GetkafkaProduct()
	if err != nil {
		panic("生成 kafka 生产者实例失败，err:" + err.Error())
		return
	}
	fmt.Println("connect kafka success !!!!!")
}

// 生成 kafka 生产者实例
func GetkafkaProduct() (sarama.SyncProducer, error) {
	config := sarama.NewConfig()
	if conf.C.Kafka.User != "" && conf.C.Kafka.Password != "" {
		config.Net.SASL.Enable = true
		config.Net.SASL.User = conf.C.Kafka.User
		config.Net.SASL.Password = conf.C.Kafka.Password
	}
	config.Producer.RequiredAcks = sarama.NoResponse          //赋值为-1：这意味着producer在follower副本确认接收到数据后才算一次发送完成。
	config.Producer.Partitioner = sarama.NewRandomPartitioner //写到随机分区中，默认设置8个分区
	config.Producer.Return.Successes = true                   // 是否等待成功后的响应,只有上面的RequiredAcks设置不是NoReponse这里才有用.
	config.Producer.Return.Errors = true                      // 是否等待失败后的响应,只有上面的RequireAcks设置不是NoReponse这里才有用.
	client, err := sarama.NewSyncProducer(strings.Split(conf.C.Kafka.Address, ","), config)
	if err != nil {
		fmt.Println("producer close err, ", err)
		return nil, err
	}
	return client, nil
}

// 生成 kafka 消费实例
func GetKafkaClient() (sarama.ConsumerGroup, context.Context, error) {
	return GetKafkaClientByGroupId(conf.C.Kafka.GroupID)
}

// 生成 kafka 消费实例
func GetKafkaClientByGroupId(groupID string) (sarama.ConsumerGroup, context.Context, error) {
	config := sarama.NewConfig()
	if conf.C.Kafka.User != "" && conf.C.Kafka.Password != "" {
		config.Net.SASL.Enable = true
		config.Net.SASL.User = conf.C.Kafka.User
		config.Net.SASL.Password = conf.C.Kafka.Password
	}
	config.Producer.Idempotent = true                             // 1.开启幂等性
	config.Producer.RequiredAcks = sarama.WaitForAll              // 开启幂等性后 acks 必须设置为 -1 即所有 isr 列表中的 broker 都ack后才ok
	config.Net.MaxOpenRequests = 1                                // 并发请求数也只能为1
	config.Consumer.Offsets.Initial = conf.C.Kafka.OffsetsInitial // 未找到组消费位移的时候从哪边开始消费
	consumerGroup, err := sarama.NewConsumerGroup(strings.Split(conf.C.Kafka.Address, ","), groupID, config)
	if err != nil {
		panic(err)
	}

	if err != nil {
		fmt.Println("connect kafka failed; err", err)
		return nil, nil, err
	}
	// 检查错误
	go func() {
		for err := range consumerGroup.Errors() {
			fmt.Println("group errors: ", err.Error())
		}
	}()
	ctx := context.Background()
	return consumerGroup, ctx, nil
}