consumergroup

Documentation

Index

• type ConsumerGroup
  • func NewConsumerGroup(cfg common.Config, name string, logger *zap.Logger, ...) (*ConsumerGroup, error)
  • func (cg *ConsumerGroup) AddHandler(topic common.Topic, hm common.MessageHandler)
  • func (cg *ConsumerGroup) Cleanup(_ sarama.ConsumerGroupSession) error
  • func (cg *ConsumerGroup) Close() error
  • func (cg *ConsumerGroup) Consume()
  • func (cg *ConsumerGroup) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error
  • func (cg *ConsumerGroup) Setup(sess sarama.ConsumerGroupSession) error
• type ConsumerGroupOption
  • func ConfigOption(cfg *sarama.Config) ConsumerGroupOption
  • func ErrFuncOption(f common.ErrFunc) ConsumerGroupOption

Types

type ConsumerGroup

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

func NewConsumerGroup

func NewConsumerGroup(
	cfg common.Config,
	name string,
	logger *zap.Logger,
	opts ...ConsumerGroupOption,
) (*ConsumerGroup, error)

func (*ConsumerGroup) AddHandler

func (cg *ConsumerGroup) AddHandler(topic common.Topic, hm common.MessageHandler)

func (*ConsumerGroup) Cleanup

func (cg *ConsumerGroup) Cleanup(_ sarama.ConsumerGroupSession) error

func (*ConsumerGroup) Close

func (cg *ConsumerGroup) Close() error

func (*ConsumerGroup) Consume

func (cg *ConsumerGroup) Consume()

func (*ConsumerGroup) ConsumeClaim

func (cg *ConsumerGroup) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error

func (*ConsumerGroup) Setup

func (cg *ConsumerGroup) Setup(sess sarama.ConsumerGroupSession) error

type ConsumerGroupOption

type ConsumerGroupOption func(cg *ConsumerGroup)

func ConfigOption

func ConfigOption(cfg *sarama.Config) ConsumerGroupOption

func ErrFuncOption

func ErrFuncOption(f common.ErrFunc) ConsumerGroupOption