consumer

Documentation

Index

• type Adapter
• type Config
• type Consumer
  • func New(initConfig *Config) (*Consumer, error)
  • func (c *Consumer) Close() chan error
  • func (c *Consumer) EnableLogging()
  • func (c *Consumer) Get() Adapter
  • func (c *Consumer) IsClosed() bool

Types

type Adapter

type Adapter interface {
	Close() error
	CommitOffsets() error
	Errors() <-chan error
	HighWaterMarks() map[string]map[int32]int64
	MarkOffset(msg *sarama.ConsumerMessage, metadata string)
	MarkOffsets(s *cluster.OffsetStash)
	MarkPartitionOffset(topic string, partition int32, offset int64, metadata string)
	Messages() <-chan *sarama.ConsumerMessage
	Notifications() <-chan *cluster.Notification
	Partitions() <-chan cluster.PartitionConsumer
	ResetOffset(msg *sarama.ConsumerMessage, metadata string)
	ResetOffsets(s *cluster.OffsetStash)
	ResetPartitionOffset(topic string, partition int32, offset int64, metadata string)
	Subscriptions() map[string][]int32
}

Adapter is the Kafka-Consumer interface

type Config

type Config struct {
	ConsumerGroup string
	ErrHandler    func(*error)
	KafkaBrokers  []string
	MsgHandler    func(*sarama.ConsumerMessage, *Consumer)
	NtfnHandler   func(*cluster.Notification)
	// Allow overwriting default sarama-config
	SaramaConfig *cluster.Config
	Topics       []string
}

Config wraps configuration for consumer

type Consumer

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

Consumer wraps sarama-cluster's consumer

func New

func New(initConfig *Config) (*Consumer, error)

New returns a configured Sarama Kafka-Consumer instance

func (*Consumer) Close

func (c *Consumer) Close() chan error

Close attempts to close the consumer, and returns any occurring errors over channel

func (*Consumer) EnableLogging

func (c *Consumer) EnableLogging()

EnableLogging logs events to console

func (*Consumer) Get

func (c *Consumer) Get() Adapter

Get returns the original Sarama Kafka consumer

func (*Consumer) IsClosed

func (c *Consumer) IsClosed() bool

IsClosed returns a bool specifying if Kafka consumer is closed