package consumer

import (
	"joybuywork/internal/app"
	"github.com/Shopify/sarama"
	"sync"
)

type GroupHandler struct {
	name    string
	mu      sync.Mutex
	closed  bool
	receive Receive
}

func newGroupHandler(receive Receive) *GroupHandler {
	return &GroupHandler{
		mu:      sync.Mutex{},
		closed:  false,
		receive: receive,
	}
}

// Cleanup 在消费者组关闭时执行清理操作
func (h *GroupHandler) Cleanup(session sarama.ConsumerGroupSession) error {
	return nil
}

// Setup 初始化消费者组
func (h *GroupHandler) Setup(session sarama.ConsumerGroupSession) error {
	return nil
}

// ConsumeClaim 消费消息
func (h *GroupHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
	for {
		select {
		case message := <-claim.Messages():
			_ = app.QueueConsumerPool.Submit(func() {
				h.receive(session, message)
			})
		case <-session.Context().Done():
			return nil
		}
	}
}

// Close 关闭消费者组处理程序
func (h *GroupHandler) Close() {
	h.mu.Lock()
	defer h.mu.Unlock()
	h.closed = true
}

// Closed 检查消费者组处理程序是否已关闭
func (h *GroupHandler) Closed() bool {
	h.mu.Lock()
	defer h.mu.Unlock()
	return h.closed
}