Consumer

Documentation

Index

• Variables
• func Consume(ctx context.Context, options Config) (sarama.PartitionConsumer, error)
• type Config
  • func (config Config) ParseCli(ctx *cli.Context) Config
• type Consumer
  • func ConsumeGroup(ctx context.Context, options Config) (*Consumer, *sync.WaitGroup, error)
  • func (consumer *Consumer) Cleanup(sarama.ConsumerGroupSession) error
  • func (consumer *Consumer) Close() error
  • func (consumer *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error
  • func (consumer *Consumer) Setup(sarama.ConsumerGroupSession) error

Variables

var CliOptions = libcli.CommonCliOptions(libcli.Kafka)

CliOptions ...

Functions

func Consume

func Consume(ctx context.Context, options Config) (sarama.PartitionConsumer, error)

Consume ...

Types

type Config

type Config struct {
	kafka.Config
	Group  string
	Topics []string
}

Config ...

func (Config) ParseCli

func (config Config) ParseCli(ctx *cli.Context) Config

ParseCli ...

type Consumer

type Consumer struct {
	Messages chan *sarama.ConsumerMessage
	// contains filtered or unexported fields
}

Consumer ...

func ConsumeGroup

func ConsumeGroup(ctx context.Context, options Config) (*Consumer, *sync.WaitGroup, error)

ConsumeGroup ...

func (*Consumer) Cleanup

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

Cleanup ...

func (*Consumer) Close

func (consumer *Consumer) Close() error

Close ...

func (*Consumer) ConsumeClaim

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

ConsumeClaim ...

func (*Consumer) Setup

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

Setup ...