consumer

Documentation

Index

• type ConsumerInput
• type KafkaConsumer
  • func NewKafkaConsumer(input ConsumerInput) *KafkaConsumer
  • func (kc *KafkaConsumer) DecodeAndPrintMsg(msg *sarama.ConsumerMessage) error
  • func (kc *KafkaConsumer) GetMsgChan() chan *sarama.ConsumerMessage
  • func (kc *KafkaConsumer) InitSaramaConsumer() error
  • func (kc *KafkaConsumer) Stop()

Types

type ConsumerInput

type ConsumerInput struct {
	// KafkaBrokers is a string of addresses of Kafka broker systems
	KafkaBrokers      []string
	KafkaVersion      sarama.KafkaVersion
	KafkaTopic        string
	KafkaProtoSchema  proto.Message
	KafkaLogErrors    bool
	MsgDelimitWithLen bool
}

type KafkaConsumer

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

func NewKafkaConsumer

func NewKafkaConsumer(input ConsumerInput) *KafkaConsumer

func (*KafkaConsumer) DecodeAndPrintMsg

func (kc *KafkaConsumer) DecodeAndPrintMsg(msg *sarama.ConsumerMessage) error

func (*KafkaConsumer) GetMsgChan

func (kc *KafkaConsumer) GetMsgChan() chan *sarama.ConsumerMessage

func (*KafkaConsumer) InitSaramaConsumer

func (kc *KafkaConsumer) InitSaramaConsumer() error

func (*KafkaConsumer) Stop

func (kc *KafkaConsumer) Stop()