kafka

Documentation

Index

• Variables
• type Consumer
  • func NewConsumer(optFns ...ConsumerOptFunc) (*Consumer, error)
  • func (c *Consumer) Cleanup(sarama.ConsumerGroupSession) error
  • func (c *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error
  • func (c *Consumer) RegisterHandler(topic string, handler handlFunc)
  • func (c *Consumer) Setup(sarama.ConsumerGroupSession) error
  • func (c *Consumer) Start(ctx context.Context) error
• type ConsumerOptFunc
  • func WithConsumerAutoCommit(autoCommit bool) ConsumerOptFunc
  • func WithConsumerBrokers(brokers []string) ConsumerOptFunc
  • func WithConsumerGroup(group string) ConsumerOptFunc
  • func WithConsumerOldest(oldest bool) ConsumerOptFunc
• type ConsumerOption
• type Producer
  • func NewProducer(optFns ...ProducerOptionFnc) (*Producer, error)
  • func (p *Producer) Close()
  • func (p *Producer) SendMessage(message *sarama.ProducerMessage) (partition int32, offset int64, err error)
• type ProducerOption
• type ProducerOptionFnc
  • func WithNumProducers(numProducers int) ProducerOptionFnc
  • func WithProducerBrokers(brokers []string) ProducerOptionFnc
  • func WithProducerTopic(topic string) ProducerOptionFnc

Variables

var DefaultConsumerOption = &ConsumerOption{
	Brokers:    []string{"localhost:9092"},
	Group:      "group_1",
	Topic:      []string{},
	Oldest:     true,
	AutoCommit: true,
}

var DefaultProducerOpt = &ProducerOption{
	Brokers:      []string{"localhost:9092"},
	NumProducers: 1,
}

Types

type Consumer

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

func NewConsumer

func NewConsumer(optFns ...ConsumerOptFunc) (*Consumer, error)

func (*Consumer) Cleanup

func (c *Consumer) Cleanup(sarama.ConsumerGroupSession) error

func (*Consumer) ConsumeClaim

func (c *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error

func (*Consumer) RegisterHandler

func (c *Consumer) RegisterHandler(topic string, handler handlFunc)

func (*Consumer) Setup

func (c *Consumer) Setup(sarama.ConsumerGroupSession) error

func (*Consumer) Start

func (c *Consumer) Start(ctx context.Context) error

type ConsumerOptFunc

type ConsumerOptFunc func(opt *ConsumerOption) *ConsumerOption

func WithConsumerAutoCommit

func WithConsumerAutoCommit(autoCommit bool) ConsumerOptFunc

func WithConsumerBrokers

func WithConsumerBrokers(brokers []string) ConsumerOptFunc

func WithConsumerGroup

func WithConsumerGroup(group string) ConsumerOptFunc

func WithConsumerOldest

func WithConsumerOldest(oldest bool) ConsumerOptFunc

type ConsumerOption

type ConsumerOption struct {
	Brokers    []string
	Group      string
	Topic      []string
	Oldest     bool
	AutoCommit bool
}

type Producer

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

func NewProducer

func NewProducer(optFns ...ProducerOptionFnc) (*Producer, error)

func (*Producer) Close

func (p *Producer) Close()

func (*Producer) SendMessage

func (p *Producer) SendMessage(message *sarama.ProducerMessage) (partition int32, offset int64, err error)

type ProducerOption

type ProducerOption struct {
	Brokers      []string
	Topic        string
	NumProducers int
}

type ProducerOptionFnc

type ProducerOptionFnc func(opt *ProducerOption) *ProducerOption

func WithNumProducers

func WithNumProducers(numProducers int) ProducerOptionFnc

func WithProducerBrokers

func WithProducerBrokers(brokers []string) ProducerOptionFnc

func WithProducerTopic

func WithProducerTopic(topic string) ProducerOptionFnc