consumer

Documentation

Index

• type Consumer
  • func New(client KafkaClient, processor processor.Processor, logger *zap.SugaredLogger, ...) *Consumer
  • func (c *Consumer) Run(ctx context.Context)
• type Fetches
• type KafkaClient
• type KgoClientAdapter
  • func (a *KgoClientAdapter) Close()
  • func (a *KgoClientAdapter) PollFetches(ctx context.Context) Fetches

Types

type Consumer

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

Consumer handles the message processing logic.

func New

func New(client KafkaClient, processor processor.Processor, logger *zap.SugaredLogger, producer *internalKgo.Producer) *Consumer

func (*Consumer) Run

func (c *Consumer) Run(ctx context.Context)

type Fetches

type Fetches interface {
	Errors() []kgo.FetchError
	EachRecord(func(*kgo.Record))
}

Fetches defines the interface for the result of a poll operation.

type KafkaClient

type KafkaClient interface {
	PollFetches(context.Context) Fetches
	Close()
}

KafkaClient defines the interface for the Kafka client operations we need.

type KgoClientAdapter

type KgoClientAdapter struct {
	Client *kgo.Client
}

KgoClientAdapter adapts the concrete *kgo.Client to our KafkaClient interface.

func (*KgoClientAdapter) Close

func (a *KgoClientAdapter) Close()

func (*KgoClientAdapter) PollFetches

func (a *KgoClientAdapter) PollFetches(ctx context.Context) Fetches