consumer

Documentation

Index

• Constants
• type KafkaConsumer
  • func NewKafkaConsumer(req *types.NewKafkaConsumerReq, ...) *KafkaConsumer
  • func (kc *KafkaConsumer) Close()
  • func (kc *KafkaConsumer) Start(ctx context.Context) (err error)
  • func (kc *KafkaConsumer) Stop()
  • func (kc *KafkaConsumer) UpdateStatus(partition *kafka.TopicPartition)
• type TopicPartMeta

Constants

const (
	KAFKA_CONSUMER_EVENT_UNINITIALIZED      = 0x00
	KAFKA_CONSUMER_EVENT_ASSIGNEDPARTITIONS = 0x10
	KAFKA_CONSUMER_EVENT_REVOKEDPARTITIONS  = 0x11
	KAFKA_CONSUMER_EVENT_MESSAGE            = 0x12
	KAFKA_CONSUMER_EVENT_PARTITIONEOF       = 0x13
	KAFKA_CONSUMER_EVENT_ERROR              = 0x14
	KAFKA_CONSUMER_EVENT_CLOSE              = 0x20
	KAFKA_CONSUMER_EVENT_INITIALIZATION     = 0x21
	KAFKA_CONSUMER_SESSION_TIMEOUT_MS       = 6000
)

Types

type KafkaConsumer

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

func NewKafkaConsumer

func NewKafkaConsumer(req *types.NewKafkaConsumerReq, dealMessageFunc func(ctx context.Context, req *types.KafkaReq) (err error), broker string) *KafkaConsumer

func (*KafkaConsumer) Close

func (kc *KafkaConsumer) Close()

func (*KafkaConsumer) Start

func (kc *KafkaConsumer) Start(ctx context.Context) (err error)

func (*KafkaConsumer) Stop

func (kc *KafkaConsumer) Stop()

func (*KafkaConsumer) UpdateStatus

func (kc *KafkaConsumer) UpdateStatus(partition *kafka.TopicPartition)

type TopicPartMeta

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