consumer

Documentation

Index

• func InitKafkaConsumer()
• func SetConsumerOption(cfg *model.KafkaConsumerConfig, opts *ConsumerOptions) error
• func SetReturnTopicOption(tag string, opts *ConsumerOptions) error
• type ConsumerOptions
  • func NewConsumerOptions() *ConsumerOptions
• type IHandler
• type SimConsumerGroup
  • func NewSimConsumerGroup(opts *ConsumerOptions) *SimConsumerGroup
  • func (scg *SimConsumerGroup) Cleanup(s sarama.ConsumerGroupSession) error
  • func (scg *SimConsumerGroup) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error
  • func (scg *SimConsumerGroup) DealErrors()
  • func (scg *SimConsumerGroup) HandleLoop()
  • func (scg *SimConsumerGroup) SetHandler(handler IHandler)
  • func (scg *SimConsumerGroup) Setup(s sarama.ConsumerGroupSession) error
  • func (scg *SimConsumerGroup) Start() error
  • func (scg *SimConsumerGroup) Stop()

Functions

func InitKafkaConsumer

func InitKafkaConsumer()

func SetConsumerOption

func SetConsumerOption(cfg *model.KafkaConsumerConfig, opts *ConsumerOptions) error

func SetReturnTopicOption

func SetReturnTopicOption(tag string, opts *ConsumerOptions) error

Types

type ConsumerOptions

type ConsumerOptions struct {
	BrokerAddresses []string
	Topics          []string
	GroupID         string
	Version         string

	Cfg *sarama.Config
}

func NewConsumerOptions

func NewConsumerOptions() *ConsumerOptions

type IHandler

type IHandler interface {
	HandleMessage(message *sarama.ConsumerMessage) error
}

type SimConsumerGroup

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

func NewSimConsumerGroup

func NewSimConsumerGroup(opts *ConsumerOptions) *SimConsumerGroup

func (*SimConsumerGroup) Cleanup

func (scg *SimConsumerGroup) Cleanup(s sarama.ConsumerGroupSession) error

func (*SimConsumerGroup) ConsumeClaim

func (scg *SimConsumerGroup) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error

func (*SimConsumerGroup) DealErrors

func (scg *SimConsumerGroup) DealErrors()

func (*SimConsumerGroup) HandleLoop

func (scg *SimConsumerGroup) HandleLoop()

func (*SimConsumerGroup) SetHandler

func (scg *SimConsumerGroup) SetHandler(handler IHandler)

func (*SimConsumerGroup) Setup

func (scg *SimConsumerGroup) Setup(s sarama.ConsumerGroupSession) error

func (*SimConsumerGroup) Start

func (scg *SimConsumerGroup) Start() error

func (*SimConsumerGroup) Stop

func (scg *SimConsumerGroup) Stop()