kafka

Documentation

Index

• Constants
• type Consumer
  • func NewConsumer(ctx context.Context, wg *sync.WaitGroup, kafkaBootstrap string, ...) (consumer *Consumer, err error)
  • func (this *Consumer) Cleanup(sarama.ConsumerGroupSession) error
  • func (this *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error
  • func (this *Consumer) Setup(sarama.ConsumerGroupSession) error

Constants

const Earliest = sarama.OffsetOldest

const Latest = sarama.OffsetNewest

Types

type Consumer

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

func NewConsumer

func NewConsumer(ctx context.Context, wg *sync.WaitGroup, kafkaBootstrap string, topics []string, groupId string, offset int64, listener func(topic string, msg []byte, time time.Time) error, errorhandler func(err error, consumer *Consumer), debug bool) (consumer *Consumer, err error)

func (*Consumer) Cleanup

func (this *Consumer) Cleanup(sarama.ConsumerGroupSession) error

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

func (*Consumer) ConsumeClaim

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

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

func (*Consumer) Setup

func (this *Consumer) Setup(sarama.ConsumerGroupSession) error