package common

import (
	"fmt"
	"github.com/Shopify/sarama"
	l4g "github.com/alecthomas/log4go"
	"strings"
	"time"
)

var consumer sarama.Consumer

// 消费者回调函数
type ConsumerCallback func(data []byte)

// 初始化消费者
func InitConsumer(hosts string) error {
	config := sarama.NewConfig()
	client, err := sarama.NewClient(strings.Split(hosts, ","), config)
	if err != nil {
		l4g.Error("unable to create kafka client: ", err)
		return err
	}

	consumer, err = sarama.NewConsumerFromClient(client)
	if err != nil {
		l4g.Error(err)
		return err
	}
	return nil
}

// 消费者循环
func LoopConsumer(topic, selectData string) (bool, error) {
	l4g.Info("topic:%s", topic)
	partitionConsumer, err := consumer.ConsumePartition(topic, 0, sarama.OffsetNewest)
	if err != nil {
		l4g.Error(err)
		return false, err
	}
	defer partitionConsumer.Close()

	for {
		select {
		case msg := <-partitionConsumer.Messages():

			l4g.Info("kafka msg is:%s", string(msg.Value))
			data := strings.Split(string(msg.Value), ":")
			msgList := strings.Split(data[9], "\"")
			l4g.Info("msgData.Sql is:%s, selectData is:%s", msgList[1], selectData)
			mysqlData := strings.Replace(strings.Replace(selectData, " ", "", -1), ";", "", -1)
			if strings.ToLower(strings.Replace(strings.Replace(msgList[1], " ", "", -1), ";", "", -1)) == strings.ToLower(mysqlData) {
				l4g.Info("msgData.Sql is:%s, selectData is:%s", strings.Replace(strings.Replace(msgList[1], " ", "", -1), ";", "", -1), strings.ToLower(mysqlData))
				return true, nil
			} else {
				return false, fmt.Errorf("downstream data is not same with upstream:%s", selectData)
			}
		case <-time.After(100 * time.Second):
			l4g.Info("Cannot receive msg in kafka\n")
			return false, fmt.Errorf("can not recieve msg:%s", selectData)

		}
	}
}

func Close() {
	if consumer != nil {
		consumer.Close()
	}
}