antenna

type Antenna ¶

type Antenna struct {
	// contains filtered or unexported fields
}

Antenna is a helper for consuming Kafka messages

func NewAntenna(
	topic string,
	brokers []string,
	groupID string,
	opts ...Option,
) *Antenna

func (a *Antenna) Run() error

func (a *Antenna) Stop()

type Handler ¶

type Handler interface {
	Process(ctx context.Context, msg kafka.Message) error
}

A Handler responds to a Kafka message

type HandlerFunc ¶

type HandlerFunc func(ctx context.Context, msg kafka.Message) error

The HandlerFunc type is an adapter to allow the use of ordinary functions as a Kafka handler. If f is a function with the appropriate signature, HandlerFunc(f) is a Handler that calls f.

func (f HandlerFunc) Process(ctx context.Context, msg kafka.Message) error

type Logger ¶

type Logger interface {
	Printf(string, ...interface{})
}

Logger interface API for log.Logger.

type LoggerFunc ¶

type LoggerFunc func(msg string, args ...any)

LoggerFunc is a bridge between Logger and any third party logger Usage:

l := NewLogger() // some logger
a := antenna.NewAntenna(...)
a.SetLogger(LoggerFunc(l.Infof))
a.SetErrorLogger(LoggerFunc(l.Errorf))

func (f LoggerFunc) Printf(msg string, args ...any)

type OffsetProvider ¶

type OffsetProvider interface {
	GetOffset(ctx context.Context, topic string, partition int) (int64, error)
}

OffsetProvider provides offset for a topic partition

type Option ¶

type Option func(*Antenna)

Option is a function that configures an Antenna.

func WithErrorLogger(logger Logger) Option

func WithLogger(logger Logger) Option

func WithMessageHandler(handler Handler) Option

func WithOffsetProvider(offsetProvider OffsetProvider) Option