consumer

Apr 5, 2024

Changes in this version

+ func DialerSCRAM512(username string, password string) (*kafka.Dialer, error)

+ func NonStopExponentialBackOff() backoff.BackOff

+ type ClientLogger interface

+ Errorf func(string, ...interface{})

+ Infof func(string, ...interface{})

+ type Config struct

+ Brokers []string

+ ID string

+ MaxBytes int

+ MaxWait time.Duration

+ MinBytes int

+ QueueCapacity int

+ Topic string

+ type Consumer struct

+ func NewConsumer(config Config, opts ...Option) *Consumer

+ func (c *Consumer) Run(ctx context.Context, handler Handler) error

+ func (c *Consumer) Stop() error

+ type Group struct

+ ID string

+ func NewGroup(config GroupConfig, opts ...Option) *Group

+ func (g *Group) Run(ctx context.Context, handler Handler) <-chan error

+ func (g *Group) Stop()

+ type GroupConfig struct

+ Brokers []string

+ Count int

+ GroupID string

+ MaxBytes int

+ MaxWait time.Duration

+ MinBytes int

+ QueueCapacity int

+ Topic string

+ type Handler func(ctx context.Context, msg Message) error

+ type HandlerRetryBackOffConstructor func() backoff.BackOff

+ type Message struct

+ type Metadata struct

+ Attempt int

+ ConsumerID string

+ GroupID string

+ type NotifyError func(ctx context.Context, err error, msg Message)

+ type Option func(consumer *Consumer)

+ func WithDataDogTracing() Option

+ func WithExplicitCommit() Option

+ func WithGroupBalancers(groupBalancers ...kafka.GroupBalancer) Option

+ func WithHandlerBackOffRetry(backOffConstructor HandlerRetryBackOffConstructor) Option

+ func WithKafkaDialer(dialer *kafka.Dialer) Option

+ func WithKafkaReader(readerFn func() Reader) Option

+ func WithLogger(logger ClientLogger) Option

+ func WithNotifyError(notifier NotifyError) Option

+ func WithQueueCapacity(queueCapacity int) Option

+ type Reader interface

+ Close func() error

+ CommitMessages func(ctx context.Context, msgs ...kafka.Message) error

+ FetchMessage func(ctx context.Context) (kafka.Message, error)

+ ReadMessage func(ctx context.Context) (kafka.Message, error)