res_levis_backend/internal/pkg/kafkaclient/consumer.go

package kafkaclient

import (
	"context"
	"encoding/json"
	"fmt"
	"log/slog"
	"sync"

	"github.com/segmentio/kafka-go"
)

func Consume[T any](r *kafka.Reader, ch chan<- T, ctx context.Context, wg *sync.WaitGroup) {
	defer wg.Done()
	for {
		select {
		case <-ctx.Done():
			msg := "consumer closed"
			slog.Info(msg)
			return
		default:
			msg, err := r.ReadMessage(ctx)
			if err != nil {
				msg := fmt.Sprintf("error reading message: %v", err)
				slog.Error(msg)
				continue
			}

			var data T
			if err := json.Unmarshal(msg.Value, &data); err != nil {
				msg := fmt.Sprintf("error decoding: %v", err)
				slog.Error(msg)
				continue
			}

			ch <- data
		}
	}
}