Feat: function that returns consumer running status

Author: asheswook

internal/queue/kafka/consumer.go

@@ -7,11 +7,13 @@ import (
 
 type Consumer struct {
 	callback func(*sarama.ConsumerMessage)
+	ready    chan bool
 }
 
 func NewConsumer(callback func(message *sarama.ConsumerMessage)) *Consumer {
 	return &Consumer{
 		callback: callback,
+		ready:    make(chan bool),
 	}
 }
 
@@ -36,6 +38,7 @@ func (c *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim saram
 
 // Setup is run at the beginning of a new session, before ConsumeClaim.
 func (c *Consumer) Setup(session sarama.ConsumerGroupSession) error {
+	close(c.ready)
 	log.Print("Consumer up and running")
 	return nil
 }

internal/queue/kafka/operator.go

@@ -64,6 +64,19 @@ func (k *ConsumeOperator[Msg]) Init() {
 	k.initialized = true
 }
 
+func (k *ConsumeOperator[Msg]) Running() bool {
+	select {
+	case _, ok := <-k.consumer.ready:
+		if !ok { // Channel 닫힘, Ready 됐다는 거
+			return true
+		}
+	default: // Channel 안 닫힘, Ready 인 됐다는 거
+		return false
+	}
+
+	return false
+}
+
 func (k *ConsumeOperator[Msg]) StartConsume() error {
 	if !k.initialized {
 		k.Init()