kafka

Documentation

Index

• func ConvertHeaderToMap(header sarama.RecordHeader) map[string]string
• type AsyncProducer
  • func NewAsyncProducer(kafkaVersion string, brokerList []string) (*AsyncProducer, error)
  • func (p *AsyncProducer) BuildProducerMessage(topicName string, key string, message string, headers []sarama.RecordHeader) *sarama.ProducerMessage
  • func (p *AsyncProducer) BuildProducerMessageHeader(key string, value string) sarama.RecordHeader
  • func (p *AsyncProducer) Close() error
  • func (p *AsyncProducer) Produce(topicName string, message interface{}) error
• type ConsumerGroup
  • func NewConsumerGroup(kafkaVersion string, brokerList []string, groupName string, initOffset int64) (cg *ConsumerGroup, err error)
  • func (cg *ConsumerGroup) Close() error
  • func (cg *ConsumerGroup) Consume(ctx context.Context, topicName string, handler sarama.ConsumerGroupHandler) (err error)
• type DefaultConsumerGroupHandler
  • func (DefaultConsumerGroupHandler) Cleanup(_ sarama.ConsumerGroupSession) error
  • func (h DefaultConsumerGroupHandler) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error
  • func (h DefaultConsumerGroupHandler) Setup(_ sarama.ConsumerGroupSession) error

Functions

func ConvertHeaderToMap

func ConvertHeaderToMap(header sarama.RecordHeader) map[string]string

Types

type AsyncProducer

type AsyncProducer struct {
	KafkaVersion sarama.KafkaVersion
	BrokerList   []string
	Config       *sarama.Config
	Client       sarama.Client
	Producer     sarama.AsyncProducer
}

func NewAsyncProducer

func NewAsyncProducer(kafkaVersion string, brokerList []string) (*AsyncProducer, error)

func (*AsyncProducer) BuildProducerMessage

func (p *AsyncProducer) BuildProducerMessage(topicName string, key string, message string, headers []sarama.RecordHeader) *sarama.ProducerMessage

func (*AsyncProducer) BuildProducerMessageHeader

func (p *AsyncProducer) BuildProducerMessageHeader(key string, value string) sarama.RecordHeader

func (*AsyncProducer) Close

func (p *AsyncProducer) Close() error

func (*AsyncProducer) Produce

func (p *AsyncProducer) Produce(topicName string, message interface{}) error

type ConsumerGroup

type ConsumerGroup struct {
	KafkaVersion sarama.KafkaVersion
	BrokerList   []string
	GroupName    string
	Config       *sarama.Config
	Client       sarama.Client
	Group        sarama.ConsumerGroup
}

func NewConsumerGroup

func NewConsumerGroup(kafkaVersion string, brokerList []string, groupName string, initOffset int64) (cg *ConsumerGroup, err error)

func (*ConsumerGroup) Close

func (cg *ConsumerGroup) Close() error

func (*ConsumerGroup) Consume

func (cg *ConsumerGroup) Consume(ctx context.Context, topicName string, handler sarama.ConsumerGroupHandler) (err error)

type DefaultConsumerGroupHandler

type DefaultConsumerGroupHandler struct{}

func (DefaultConsumerGroupHandler) Cleanup

func (DefaultConsumerGroupHandler) Cleanup(_ sarama.ConsumerGroupSession) error

func (DefaultConsumerGroupHandler) ConsumeClaim

func (h DefaultConsumerGroupHandler) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error

func (DefaultConsumerGroupHandler) Setup

func (h DefaultConsumerGroupHandler) Setup(_ sarama.ConsumerGroupSession) error