consumer

Documentation

Index

• type BatchConfiguration
• type BatchConsumeFn
• type BatchConsumer
  • func NewBatchConsumer(name string) BatchConsumer
  • func NewBatchConsumerFn(name string, fn BatchConsumeFn) BatchConsumer
• type BatchConsumerInterceptor
• type BatchConsumerInterceptorFunc
  • func (f BatchConsumerInterceptorFunc) Intercept(messages []*Message, next BatchConsumeFn) error
• type BatchConsumerStarter
  • func (s *BatchConsumerStarter) Start() error
  • func (s *BatchConsumerStarter) Stop() error
• type ConsumeFn
• type Consumer
  • func New(name string) Consumer
  • func NewFn(name string, fn ConsumeFn) Consumer
• type Interceptor
• type InterceptorFunc
  • func (f InterceptorFunc) Intercept(msg *Message, next ConsumeFn) error
• type Message
• type Messages
• type RetryConfiguration
• type SingleConsumerStarter
  • func (s *SingleConsumerStarter) Start() error
  • func (s *SingleConsumerStarter) Stop() error
• type Starter
  • func NewBatchConsumerStarter(cfg *config.Config, c BatchConsumer, interceptors []BatchConsumerInterceptor) (Starter, error)
  • func NewStarter(cfg *config.Config, c Consumer, interceptors []Interceptor) (Starter, error)

Types

type BatchConfiguration

type BatchConfiguration struct {
	MessageGroupLimit         int
	MessageGroupByteSizeLimit any
}

type BatchConsumeFn

type BatchConsumeFn = kafkalib.BatchConsumeFn

type BatchConsumer

type BatchConsumer interface {
	Name() string
	Consume([]*Message) error
}

func NewBatchConsumer

func NewBatchConsumer(name string) BatchConsumer

func NewBatchConsumerFn

func NewBatchConsumerFn(name string, fn BatchConsumeFn) BatchConsumer

type BatchConsumerInterceptor

type BatchConsumerInterceptor interface {
	Intercept(msg []*Message, next BatchConsumeFn) error
}

type BatchConsumerInterceptorFunc

type BatchConsumerInterceptorFunc func(msg []*Message, next BatchConsumeFn) error

func (BatchConsumerInterceptorFunc) Intercept

func (f BatchConsumerInterceptorFunc) Intercept(messages []*Message, next BatchConsumeFn) error

type BatchConsumerStarter

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

func (*BatchConsumerStarter) Start

func (s *BatchConsumerStarter) Start() error

func (*BatchConsumerStarter) Stop

func (s *BatchConsumerStarter) Stop() error

type ConsumeFn

type ConsumeFn = kafkalib.ConsumeFn

type Consumer

type Consumer interface {
	Name() string
	Consume(*Message) error
}

Consumer represent consumer func with name, so consumer configuration can be matched by using name

func New

func New(name string) Consumer

func NewFn

func NewFn(name string, fn ConsumeFn) Consumer

type Interceptor

type Interceptor interface {
	Intercept(msg *Message, next ConsumeFn) error
}

Interceptor represent a middleware for consume functions

type InterceptorFunc

type InterceptorFunc func(msg *Message, next ConsumeFn) error

InterceptorFunc is a functional implementation of Interceptor

func (InterceptorFunc) Intercept

func (f InterceptorFunc) Intercept(msg *Message, next ConsumeFn) error

type Message

type Message = kafkalib.Message

type Messages

type Messages = []*kafkalib.Message

type RetryConfiguration

type RetryConfiguration struct {
	ClientID        string
	StartTimeCron   string
	Topic           string
	DeadLetterTopic string
	Rack            string
	Brokers         []string
	MaxRetry        int
	WorkDuration    int
}

type SingleConsumerStarter

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

func (*SingleConsumerStarter) Start

func (s *SingleConsumerStarter) Start() error

func (*SingleConsumerStarter) Stop

func (s *SingleConsumerStarter) Stop() error

type Starter

type Starter interface {
	Start() error
	Stop() error
}

func NewBatchConsumerStarter

func NewBatchConsumerStarter(cfg *config.Config, c BatchConsumer, interceptors []BatchConsumerInterceptor) (Starter, error)

func NewStarter

func NewStarter(cfg *config.Config, c Consumer, interceptors []Interceptor) (Starter, error)