service/kafka.go

package service

import (
	"github.com/confluentinc/confluent-kafka-go/v2/kafka"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/client_golang/prometheus/promauto"
)

type ConsumerHandler func(*kafka.Message) error
type ConsumerOptions struct {
	Counter prometheus.Counter
}

func ConsumerOptionsWithCounter() ConsumerOptions {
	return ConsumerOptions{
		Counter: promauto.NewCounter(prometheus.CounterOpts{
			Namespace: "kafka",
			Subsystem: "consumer",
			Name:      "received",
			Help:      "The total number of received messages",
		}),
	}
}

type Consumer interface {
	Service

	SetPause(bool) error
	Paused() bool

	Subscribe([]string, chan *kafka.Message, ConsumerOptions) error
	Subscribed() bool

	RegisterHandlers(...ConsumerHandler)
}

type Producer interface {
	Service
	Produce(*kafka.Message) error
}