/**
 * @api post kafka.
 *
 * User: yunshengzhu
 * Date: 2022/3/16
 * Time: 下午10:09
 */
package kafka

import (
	"context"
	"gitee.com/zhuyuns/basic/logs"
	"github.com/Shopify/sarama"
)

type consumerGroupHandler struct {
	name string
	fn   func(ms *sarama.ConsumerMessage) error
}

func (consumerGroupHandler) Setup(_ sarama.ConsumerGroupSession) error   { return nil }
func (consumerGroupHandler) Cleanup(_ sarama.ConsumerGroupSession) error { return nil }
func (h consumerGroupHandler) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
	for msg := range claim.Messages() {
		err := h.fn(msg)
		if err != nil {
			logs.Log.Info("业务处理失败,数据未提交")
			return err
		}
		sess.MarkMessage(msg, "")
	}
	return nil
}

type ConsumeGroup struct {
	client sarama.Client
}

func (c *ConsumeGroup) Consume(groupId string, topics []string, fn func(msg *sarama.ConsumerMessage) error) error {
	group, err := sarama.NewConsumerGroupFromClient(groupId, c.client)
	if err != nil {
		return err
	}
	ctx := context.Background()
	for {
		handler := consumerGroupHandler{name: groupId, fn: fn}
		err := group.Consume(ctx, topics, handler)
		if err != nil {
			return err
		}
	}
}

func (c *ConsumeGroup) Close() error {
	return c.client.Close()
}

func NewConsumeGroup(kafkaAddress []string) (*ConsumeGroup, error) {
	kafkaConfig := sarama.NewConfig()
	kafkaConfig.Consumer.Return.Errors = false
	kafkaConfig.Version = sarama.V0_10_2_0
	client, err := sarama.NewClient(kafkaAddress, kafkaConfig)
	if err != nil {
		return nil, err
	}
	return &ConsumeGroup{
		client: client,
	}, nil
}