package Test

import (
	"fmt"
	"github.com/Shopify/sarama"
	"github.com/sirupsen/logrus"
	"github.com/tracer0tong/kafkalogrus"
	"testing"
	"time"
)

var KafkaAddress = "10.10.6.202:9092"
var KafkaTopic = "dsLogBase"

func TestLogrusKafka(t *testing.T) {
	//writeKafka()
	readKafka()
}
func writeKafka() {
	var err error
	var hook *kafkalogrus.KafkaLogrusHook
	if hook, err = kafkalogrus.NewKafkaLogrusHook(
		"kh",
		logrus.AllLevels,
		&logrus.JSONFormatter{},
		[]string{KafkaAddress},
		KafkaTopic,
		true, nil); err != nil {
		panic(err)
	}
	logger := logrus.New()
	logger.Hooks.Add(hook)
	l := logger.WithField("topic", KafkaTopic)
	l.Debug("This must not be logged")
	l.Info("This is an Info msg")
	l.Warn("This is a Warn msg")
	l.Error("This is an Error msg")
	time.Sleep(time.Second)
}

func readKafka() {
	fmt.Printf("consumer_test")
	config := sarama.NewConfig()
	config.Consumer.Return.Errors = true
	config.Version = sarama.V0_11_0_2
	// consumer
	consumer, err := sarama.NewConsumer([]string{KafkaAddress}, config)
	if err != nil {
		fmt.Printf("consumer_test create consumer error %s\n", err.Error())
		return
	}
	defer consumer.Close()
	partitionConsumer, err := consumer.ConsumePartition(KafkaTopic, 0, sarama.OffsetOldest)
	if err != nil {
		fmt.Printf("try create partition_consumer error %s\n", err.Error())
		return
	}
	defer partitionConsumer.Close()
	for {
		select {
		case msg := <-partitionConsumer.Messages():
			fmt.Printf("msg offset: %d, partition: %d, timestamp: %s, value: %s\n",
				msg.Offset, msg.Partition, msg.Timestamp.String(), string(msg.Value))
		case err := <-partitionConsumer.Errors():
			fmt.Printf("err :%s\n", err.Error())
		}
	}
	//注意：需要自行记录partition和offset才能保证不重复消费数据！用一个文件来记录吧！
}