Why Go
- Case Studies
  
  Common problems companies solve with Go
- Use Cases
  
  Stories about how and why companies use Go
- Security Policy
  
  How Go can help keep you secure by default
Learn
Docs
- Effective Go
  
  Tips for writing clear, performant, and idiomatic Go code
- Go User Manual
  
  A complete introduction to building software with Go
- Standard library
  
  Reference documentation for Go's standard library
- Release Notes
  
  Learn what's new in each Go release
Packages
Community
- Recorded Talks
  
  Videos from prior events
- Meetups
  
  Meet other local Go developers
- Conferences
  
  Learn and network with Go developers from around the world
- Go blog
  
  The Go project's official blog.
- Go project
  
  Get help and stay informed from Go
- Get connected

kafka

package module

Go to main page

Versions in this module

v1

v1.5.0

Feb 2, 2018

Changes in this version

type BrokerConf

+ TLSCa []byte

+ TLSCert []byte

+ TLSKey []byte

v0

v0.10.0

Sep 21, 2016

Changes in this version

+ const StartOffsetNewest

+ const StartOffsetOldest

+ var ErrClosed = errors.New("closed")

+ var ErrMxClosed = errors.New("closed")

+ var ErrNoData = errors.New("no data")

+ type BatchConsumer interface

+ ConsumeBatch func() ([]*proto.Message, error)

+ type Broker struct

+ func Dial(nodeAddresses []string, conf BrokerConf) (*Broker, error)

+ func (b *Broker) BatchConsumer(conf ConsumerConf) (BatchConsumer, error)

+ func (b *Broker) Close()

+ func (b *Broker) Consumer(conf ConsumerConf) (Consumer, error)

+ func (b *Broker) Metadata() (*proto.MetadataResp, error)

+ func (b *Broker) OffsetCoordinator(conf OffsetCoordinatorConf) (OffsetCoordinator, error)

+ func (b *Broker) OffsetEarliest(topic string, partition int32) (offset int64, err error)

+ func (b *Broker) OffsetLatest(topic string, partition int32) (offset int64, err error)

+ func (b *Broker) PartitionCount(topic string) (int32, error)

+ func (b *Broker) Producer(conf ProducerConf) Producer

+ type BrokerConf struct

+ AllowTopicCreation bool

+ ClientID string

+ DialRetryLimit int

+ DialRetryWait time.Duration

+ DialTimeout time.Duration

+ LeaderRetryLimit int

+ LeaderRetryWait time.Duration

+ Log interface{ ... }

+ Logger Logger

+ ReadTimeout time.Duration

+ RetryErrLimit int

+ RetryErrWait time.Duration

+ func NewBrokerConf(clientID string) BrokerConf

+ type Client interface

+ Close func()

+ Consumer func(conf ConsumerConf) (Consumer, error)

+ OffsetCoordinator func(conf OffsetCoordinatorConf) (OffsetCoordinator, error)

+ OffsetEarliest func(topic string, partition int32) (offset int64, err error)

+ OffsetLatest func(topic string, partition int32) (offset int64, err error)

+ Producer func(conf ProducerConf) Producer

+ type Consumer interface

+ Consume func() (*proto.Message, error)

+ type ConsumerConf struct

+ Logger Logger

+ MaxFetchSize int32

+ MinFetchSize int32

+ Partition int32

+ RequestTimeout time.Duration

+ RetryErrLimit int

+ RetryErrWait time.Duration

+ RetryLimit int

+ RetryWait time.Duration

+ StartOffset int64

+ Topic string

+ func NewConsumerConf(topic string, partition int32) ConsumerConf

+ type DistributingProducer interface

+ Distribute func(topic string, messages ...*proto.Message) (offset int64, err error)

+ func NewHashProducer(p Producer, numPartitions int32) DistributingProducer

+ func NewRandomProducer(p Producer, numPartitions int32) DistributingProducer

+ func NewRoundRobinProducer(p Producer, numPartitions int32) DistributingProducer

+ type Logger interface

+ Debug func(msg string, args ...interface{})

+ Error func(msg string, args ...interface{})

+ Info func(msg string, args ...interface{})

+ Warn func(msg string, args ...interface{})

+ type Mx struct

+ func Merge(consumers ...Consumer) *Mx

+ func (p *Mx) Close()

+ func (p *Mx) Consume() (*proto.Message, error)

+ func (p *Mx) Workers() int

+ type OffsetCoordinator interface

+ Commit func(topic string, partition int32, offset int64) error

+ Offset func(topic string, partition int32) (offset int64, metadata string, err error)

+ type OffsetCoordinatorConf struct

+ ConsumerGroup string

+ Logger Logger

+ RetryErrLimit int

+ RetryErrWait time.Duration

+ func NewOffsetCoordinatorConf(consumerGroup string) OffsetCoordinatorConf

+ type Producer interface

+ Produce func(topic string, partition int32, messages ...*proto.Message) (offset int64, err error)

+ type ProducerConf struct

+ Compression proto.Compression

+ Logger Logger

+ RequestTimeout time.Duration

+ RequiredAcks int16

+ RetryLimit int

+ RetryWait time.Duration

+ func NewProducerConf() ProducerConf

Incompatible versions in this module

v2.0.7+incompatible

May 21, 2019

v2.0.6+incompatible

May 16, 2019

v2.0.5+incompatible

Jan 21, 2019

v2.0.4+incompatible

Aug 28, 2018

v2.0.3+incompatible

Jun 26, 2018

v2.0.2+incompatible

Apr 27, 2018

v2.0.1+incompatible

Apr 17, 2018

v2.0.0+incompatible

Apr 17, 2018

Other modules containing this package

github.com/optiopay/kafka/v2