kafkaelasticsearch

Documentation

Index

• func NewConsumerGroup(brokerList []string, groupName, clientID string, useLog bool) (sarama.ConsumerGroup, error)
• func NewRepoConsumer(opt InitOption) repo.Repository
• type BulkAttr
• type ConsumerRepo
  • func (repo *ConsumerRepo) Cleanup(session sarama.ConsumerGroupSession) error
  • func (repo *ConsumerRepo) Consume(topic string)
  • func (repo *ConsumerRepo) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error
  • func (repo *ConsumerRepo) Setup(session sarama.ConsumerGroupSession) error
• type InitOption

Functions

func NewConsumerGroup

func NewConsumerGroup(brokerList []string, groupName, clientID string, useLog bool) (sarama.ConsumerGroup, error)

NewConsumerGroup create new object of consumer

func NewRepoConsumer

func NewRepoConsumer(opt InitOption) repo.Repository

NewRepoConsumer new repo consumer with kafka and es adapter

Types

type BulkAttr

type BulkAttr struct {
	Min             int
	Max             int
	Hop             int
	WaitTimeMS      int
	WaitTimeTotalMS int
	// contains filtered or unexported fields
}

BulkAttr bulk attribute

type ConsumerRepo

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

ConsumerRepo repo producer with kafka & elastic adapter

func (*ConsumerRepo) Cleanup

func (repo *ConsumerRepo) Cleanup(session sarama.ConsumerGroupSession) error

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

func (*ConsumerRepo) Consume

func (repo *ConsumerRepo) Consume(topic string)

Consume consumer streaming data from kafka store it to es will loop forever

func (*ConsumerRepo) ConsumeClaim

func (repo *ConsumerRepo) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error

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

func (*ConsumerRepo) Setup

func (repo *ConsumerRepo) Setup(session sarama.ConsumerGroupSession) error

Setup is run at the beginning of a new session, before ConsumeClaim

type InitOption

type InitOption struct {
	EsClient   *elastic.Client
	Index      string
	TypeIndex  string
	Bulk       BulkAttr
	BrokerList []string
	GroupName  string
	UseLog     bool
}

InitOption option to init repository