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
• type Producer
  • func NewProducer(conf config.Config, ctx context.Context, wg *sync.WaitGroup) (*Producer, error)
  • func (producer *Producer) Publish(topic string, msg string) 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, commitMessages 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

type Producer

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

func NewProducer

func NewProducer(conf config.Config, ctx context.Context, wg *sync.WaitGroup) (*Producer, error)

func (*Producer) Publish

func (producer *Producer) Publish(topic string, msg string) error