consumer

Documentation

Index

• type Config
• type Consumer
  • func New(config Config) Consumer
• type DefaultConsumerConfig
  • func (c DefaultConsumerConfig) ConsumerGroup() string
  • func (c DefaultConsumerConfig) InitialOffset() int64
  • func (c DefaultConsumerConfig) ProcessingTimeout() time.Duration
  • func (c DefaultConsumerConfig) Topics() []string
  • func (c DefaultConsumerConfig) ZookeeperChroot() string
  • func (c DefaultConsumerConfig) ZookeeperURL() string
• type Message

Types

type Config

type Config interface {
	ConsumerGroup() string
	InitialOffset() int64
	ProcessingTimeout() time.Duration
	ZookeeperURL() string
	ZookeeperChroot() string
	Topics() []string
}

Config represents the configuration required for a consumer service

type Consumer

type Consumer interface {
	Start() chan Message
	Commit(to Message) error
}

Consumer ...

func New

func New(config Config) Consumer

New ...

type DefaultConsumerConfig

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

DefaultConsumerConfig is a default Config implementation

func (DefaultConsumerConfig) ConsumerGroup

func (c DefaultConsumerConfig) ConsumerGroup() string

ConsumerGroup implements Config.ConsumerGroup

func (DefaultConsumerConfig) InitialOffset

func (c DefaultConsumerConfig) InitialOffset() int64

InitialOffset implements Config.InitialOffset

func (DefaultConsumerConfig) ProcessingTimeout

func (c DefaultConsumerConfig) ProcessingTimeout() time.Duration

ProcessingTimeout implements Config.ProcessingTimeout

func (DefaultConsumerConfig) Topics

func (c DefaultConsumerConfig) Topics() []string

Topics implements Config.Topics

func (DefaultConsumerConfig) ZookeeperChroot

func (c DefaultConsumerConfig) ZookeeperChroot() string

ZookeeperChroot implements Config.ZookeeperChroot

func (DefaultConsumerConfig) ZookeeperURL

func (c DefaultConsumerConfig) ZookeeperURL() string

ZookeeperURL implements Config.ZookeeperURL

type Message

type Message interface {
	Key() []byte
	Value() []byte
	Partition() int32
	Offset() int64
}

Message ...