res_levis_backend/internal/pkg/kafkaclient/consumer.go

package kafkaclient

import (
	"context"
	"encoding/json"
	"fmt"

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

func Consume[T any](r *kafka.Reader, ch chan<- T) {
	for {
		msg, err := r.ReadMessage(context.Background())
		if err != nil {
			fmt.Println("error reading message:", err)
			continue
		}

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

		ch <- data
	}
}