async_kafka

Documentation

Index

• type Committer
  • func (c *Committer) Commit(msg *kafka.Message)
  • func (c *Committer) Start()
  • func (c *Committer) Stop()
  • func (c *Committer) WaitCommits()
• type ConsumeCallback
• type Consumer
  • func NewConsumer(brokers string, groupId string, topic string, threads int) (*Consumer, error)
  • func NewConsumerWithRebalanceCb(brokers string, groupId string, topic string, threads int, ...) (*Consumer, error)
  • func NewSingleThreadConsumer(brokers string, groupId string, topic string) (*Consumer, error)
  • func (c *Consumer) Close() []error
  • func (c *Consumer) Consume(cb ConsumeCallback) []error
  • func (c *Consumer) Stop()
• type Producer
  • func NewProducer(brokers string, topic string) (*Producer, error)
  • func NewProducerWithConfig(brokers string, topic string, config kafka.ConfigMap) (*Producer, error)
  • func (p *Producer) Close()
  • func (p *Producer) Errors() chan error
  • func (p *Producer) Produce(message string) error

Types

type Committer

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

func (*Committer) Commit

func (c *Committer) Commit(msg *kafka.Message)

func (*Committer) Start

func (c *Committer) Start()

func (*Committer) Stop

func (c *Committer) Stop()

func (*Committer) WaitCommits

func (c *Committer) WaitCommits()

type ConsumeCallback

type ConsumeCallback func(msg *kafka.Message, thread int) error

type Consumer

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

func NewConsumer

func NewConsumer(brokers string, groupId string, topic string, threads int) (*Consumer, error)

func NewConsumerWithRebalanceCb

func NewConsumerWithRebalanceCb(brokers string, groupId string, topic string, threads int, cb kafka.RebalanceCb) (*Consumer, error)

func NewSingleThreadConsumer

func NewSingleThreadConsumer(brokers string, groupId string, topic string) (*Consumer, error)

func (*Consumer) Close

func (c *Consumer) Close() []error

func (*Consumer) Consume

func (c *Consumer) Consume(cb ConsumeCallback) []error

func (*Consumer) Stop

func (c *Consumer) Stop()

type Producer

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

func NewProducer

func NewProducer(brokers string, topic string) (*Producer, error)

func NewProducerWithConfig

func NewProducerWithConfig(brokers string, topic string, config kafka.ConfigMap) (*Producer, error)

func (*Producer) Close

func (p *Producer) Close()

func (*Producer) Errors

func (p *Producer) Errors() chan error

func (*Producer) Produce

func (p *Producer) Produce(message string) error