kafka

Documentation

Index

• type Consumer
  • func NewConsumer(config *sarama.Config, logger *log.Logger, topic string, groupID string, ...) *Consumer
  • func (c Consumer) CloseConnection()
  • func (c Consumer) Consume()
• type ConsumerGroupHandler
  • func (ConsumerGroupHandler) Cleanup(_ sarama.ConsumerGroupSession) error
  • func (h ConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error
  • func (ConsumerGroupHandler) Setup(_ sarama.ConsumerGroupSession) error
• type Producer
  • func NewProducer(config *sarama.Config, logger *log.Logger, topic string, brokers []string) *Producer
  • func (p *Producer) Publish(message string)

Types

type Consumer

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

func NewConsumer

func NewConsumer(config *sarama.Config, logger *log.Logger, topic string, groupID string, brokers []string) *Consumer

func (Consumer) CloseConnection

func (c Consumer) CloseConnection()

func (Consumer) Consume

func (c Consumer) Consume()

type ConsumerGroupHandler

type ConsumerGroupHandler struct{}

ConsumerGroupHandler represents the sarama consumer group

func (ConsumerGroupHandler) Cleanup

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

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

func (ConsumerGroupHandler) ConsumeClaim

func (h ConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error

ConsumeClaim must start a consumer loop of ConsumerGroupClaim's Messages(), here is supposed to be what you want to do with the message. In this example the message will be logged with the topic name, partition and message value.

func (ConsumerGroupHandler) Setup

func (ConsumerGroupHandler) Setup(_ sarama.ConsumerGroupSession) error

Setup is run before consumer start consuming, is normally used to setup things such as database connections

type Producer

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

func NewProducer

func NewProducer(config *sarama.Config, logger *log.Logger, topic string, brokers []string) *Producer

func (*Producer) Publish

func (p *Producer) Publish(message string)