res_levis_backend/internal/pkg/kafkaclient/consumer.go

package kafkaclient

import (
	"context"
	"encoding/json"
	"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 {
				slog.Error("error reading message", "error", err)
				continue
			}

			var data T
			if err := json.Unmarshal(msg.Value, &data); err != nil {

				slog.Error("error decoding", "error", err)
				continue
			}

			ch <- data
		}
	}
}