kafka

Documentation

Index

• type Options
• type Queue
  • func New(opts Options) *Queue
  • func (q *Queue) Cleanup(sarama.ConsumerGroupSession) error
  • func (q *Queue) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error
  • func (q *Queue) Run()
  • func (q *Queue) Setup(sarama.ConsumerGroupSession) error

Types

type Options

type Options struct {
	queue.Options
	Brokers      []string
	GroupID      string
	KafkaVersion string
}

Options struct

type Queue

type Queue struct {
	*queue.Base
	Options
	// contains filtered or unexported fields
}

Queue struct

func New

func New(opts Options) *Queue

New return queue base kafka

func (*Queue) Cleanup

func (q *Queue) Cleanup(sarama.ConsumerGroupSession) error

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

func (*Queue) ConsumeClaim

func (q *Queue) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error

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

func (*Queue) Run

func (q *Queue) Run()

Run use for queue run

func (*Queue) Setup

func (q *Queue) Setup(sarama.ConsumerGroupSession) error

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