input

Documentation

Index

• func ConsumerHealthCheck(consumerName string) healthcheck.Check
• type Consumer
  • func (consumer *Consumer) Cleanup(session sarama.ConsumerGroupSession) error
  • func (consumer *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error
  • func (consumer *Consumer) Setup(session sarama.ConsumerGroupSession) error
• type ConsumerError
• type Kafka
  • func NewKafka() *Kafka
  • func (k *Kafka) Description() string
  • func (k *Kafka) GetName() string
  • func (k *Kafka) Init() error
  • func (k *Kafka) Msgs() chan []byte
  • func (k *Kafka) Start() error
  • func (k *Kafka) Stop() error

Functions

func ConsumerHealthCheck

func ConsumerHealthCheck(consumerName string) healthcheck.Check

Types

type Consumer

type Consumer struct {
	Name string
	// contains filtered or unexported fields
}

Consumer represents a Sarama consumer group consumer

func (*Consumer) Cleanup

func (consumer *Consumer) Cleanup(session sarama.ConsumerGroupSession) error

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

func (*Consumer) ConsumeClaim

func (consumer *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error

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

func (*Consumer) Setup

func (consumer *Consumer) Setup(session sarama.ConsumerGroupSession) error

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

type ConsumerError

type ConsumerError struct {
	UnixTime int64
	Error    error
}

type Kafka

type Kafka struct {
	Name          string
	Version       string
	Earliest      bool
	Brokers       string
	ConsumerGroup string
	Topic         string

	Sasl struct {
		Username string
		Password string
	}
	// contains filtered or unexported fields
}

Kafka reader configuration

func NewKafka

func NewKafka() *Kafka

NewKafka get instance of kafka reader

func (*Kafka) Description

func (k *Kafka) Description() string

Description of this kafka consumre, which topic it reads from

func (*Kafka) GetName

func (k *Kafka) GetName() string

GetName name of this kafka consumer instance

func (*Kafka) Init

func (k *Kafka) Init() error

Init Initialise the kafka instance with configuration

func (*Kafka) Msgs

func (k *Kafka) Msgs() chan []byte

Msgs returns the message from kafka

func (*Kafka) Start

func (k *Kafka) Start() error

Start start kafka consumer, uses saram library

func (*Kafka) Stop

func (k *Kafka) Stop() error

Stop kafka consumer and close all connections