consumer

Documentation

Index

• type APP
• type Consumer
  • func NewConsumer(brokers, group, topics string, app APP, wg *sync.WaitGroup) (*Consumer, sarama.ConsumerGroup)
  • func (consumer *Consumer) Cleanup(sess sarama.ConsumerGroupSession) error
  • func (consumer *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error
  • func (consumer *Consumer) ProcessMessage(topic, msg string)
  • func (consumer *Consumer) SetBuffer(buf chan []string)
  • func (consumer *Consumer) Setup(sess sarama.ConsumerGroupSession) error

Types

type APP

type APP interface {
	ProcessTopic(topic, msg string, reqId int64) (maxRetryCnt int, err error)
	ProcessRetryFailure(topic, msg string, reqId int64) error
	ProcessDeadLetterTopic(topic, msg string, reqId int64) error
	Context() context.Context
	//for consumer goroutine to check exit
	CheckDone() bool
}

type Consumer

type Consumer struct {
	Ready chan bool
	// contains filtered or unexported fields
}

Consumer represents a Sarama consumer group consumer

func NewConsumer

func NewConsumer(brokers, group, topics string, app APP, wg *sync.WaitGroup) (*Consumer, sarama.ConsumerGroup)

func (*Consumer) Cleanup

func (consumer *Consumer) Cleanup(sess sarama.ConsumerGroupSession) error

Cleanup is run at the end of a session, once all ConsumeClaim goroutines have exited

func (*Consumer) ConsumeClaim

func (consumer *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error

ConsumeClaim must start a consumer loop of ConsumerGroupClaim's Messages().

func (*Consumer) ProcessMessage

func (consumer *Consumer) ProcessMessage(topic, msg string)

message protocol: [topic type][retry count][reqID][req]

func (*Consumer) SetBuffer

func (consumer *Consumer) SetBuffer(buf chan []string)

This method used to set a buffer for consumer to put retry/dead letter/result message. structure of buffer member: [topic][reqID][message] message protocol: [topic type][retry count][reqID][req]

func (*Consumer) Setup

func (consumer *Consumer) Setup(sess sarama.ConsumerGroupSession) error

Setup is run at the beginning of a new session, before ConsumeClaim