package KafkaUtil

import (
	"bytes"
	"context"
	"dsDataex/Utils/ConfigUtil"
	"fmt"
	"github.com/segmentio/kafka-go"
	"strconv"
	"time"
)

var KafkaClient *kafka.Conn
var KafkaBroker string
var CTX context.Context

func init() {

	CTX=context.Background()

	KafkaClient, _ = kafka.DialLeader(CTX , "tcp", ConfigUtil.KafkaBrokers[0], "__consumer_offsets" , 0)

	brokers,_:=	KafkaClient.Brokers()

	KafkaBroker=brokers[0].Host+":"+strconv.Itoa(brokers[0].Port)+"【"+strconv.Itoa(brokers[0].ID)+"】"
}

func ProvideLow(topic string)  {

	KafkaClient, _ = kafka.DialLeader(context.Background(), "tcp", ConfigUtil.KafkaBrokers[0], topic , 0)

	KafkaClient.SetWriteDeadline(time.Now().Add(5 *time.Second))

	KafkaClient.WriteMessages(
		kafka.Message{Value: []byte("one!")},
		kafka.Message{Value: []byte("two!")},
		kafka.Message{Value: []byte("three!")},
	)

	//KafkaClient.Close()
}

func ConsumeLow(topic string) {

	KafkaClient, _ = kafka.DialLeader(context.Background(), "tcp", ConfigUtil.KafkaBrokers[0], topic , 0)

	KafkaClient.SetReadDeadline(time.Now().Add(5 *time.Second))

	batch := KafkaClient.ReadBatch(10e3, 1e6) // fetch 10KB min, 1MB max

	for {
		b := make([]byte, 10e3) // 10KB max per message

		_, err := batch.Read(b)

		if err != nil {
			break
		}
		index := bytes.IndexByte(b, 0)

		fmt.Println(string(b[0:index]))
	}

	batch.Close()
	//KafkaClient.Close()
}

func Consume(topic string,group string) {

	r := kafka.NewReader(kafka.ReaderConfig{
		Brokers:   ConfigUtil.KafkaBrokers ,
		Topic:     topic,
		//Partition: 0,
		GroupID:   group,//必须指定 Group，否则需要指定 Partition！！！
		MinBytes:  10e3, // 10KB
		MaxBytes:  10e6, // 10MB
	})

	for {
		m, err := r.ReadMessage( CTX )

		if err != nil {
			break
		}

		fmt.Printf("message at partiton %d offset %d: %s ==> %s\n",m.Partition, m.Offset, string(m.Key), string(m.Value))
	}

	r.Close()
}

func Provide(topic string){

	w := kafka.NewWriter(kafka.WriterConfig{
		Brokers: ConfigUtil.KafkaBrokers,
		Topic:  topic,
		Balancer: &kafka.LeastBytes{},//Partition 自动分配器
	})

	w.WriteMessages(context.Background(),
		kafka.Message{
			//Key:   []byte("Key-A"),
			Value: []byte("Hello World! One"),
		},
		kafka.Message{
			Value: []byte("Hello World! Two"),
		},
		kafka.Message{
			Value: []byte("Hello World! Three"),
		},
	)

	w.Close()
}

func CreateTopic(topic string){

	KafkaClient, _ = kafka.DialLeader(CTX , "tcp", ConfigUtil.KafkaBrokers[0], "__consumer_offsets" , 0)

	err:= KafkaClient.CreateTopics(kafka.TopicConfig{
		NumPartitions: 8,
		ReplicationFactor: int(ConfigUtil.KafkaReply),
		Topic: topic,
	})

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

func DeleteTopic(topic string){
	KafkaClient, _ = kafka.DialLeader( CTX, "tcp", ConfigUtil.KafkaBrokers[0], topic , 0)

	err:= KafkaClient.DeleteTopics(topic)

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

}