consumer

Documentation

Index

• type Consumer
  • func NewConsumer(brokers, topic []string, groupName string, instanceId string) (*Consumer, error)
  • func (c *Consumer) Close() error
  • func (c *Consumer) GetMessage() *shared.KafkaMessage
  • func (c *Consumer) GetMessages() chan *shared.KafkaMessage
  • func (c *Consumer) GetStats() (uint64, uint64)
  • func (c *Consumer) IsRunning() bool
  • func (c *Consumer) MarkMessage(msg *shared.KafkaMessage)
  • func (c *Consumer) MarkMessages(msgs []*shared.KafkaMessage)
  • func (c *Consumer) Start(ctx context.Context) error
• type GroupHandler
  • func (c *GroupHandler) Cleanup(session sarama.ConsumerGroupSession) error
  • func (c *GroupHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error
  • func (c *GroupHandler) Setup(_ sarama.ConsumerGroupSession) error
• type TopicPartition

Types

type Consumer

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

Consumer wraps sarama's ConsumerGroup.

func NewConsumer

func NewConsumer(brokers, topic []string, groupName string, instanceId string) (*Consumer, error)

NewConsumer initializes a Consumer.

func (*Consumer) Close

func (c *Consumer) Close() error

Close terminates the Consumer.

func (*Consumer) GetMessage

func (c *Consumer) GetMessage() *shared.KafkaMessage

GetMessage receives a single message.

func (*Consumer) GetMessages

func (c *Consumer) GetMessages() chan *shared.KafkaMessage

GetMessages returns the message channel.

func (*Consumer) GetStats

func (c *Consumer) GetStats() (uint64, uint64)

GetStats returns marked and consumed message counts.

func (*Consumer) IsRunning

func (c *Consumer) IsRunning() bool

IsRunning returns the run state.

func (*Consumer) MarkMessage

func (c *Consumer) MarkMessage(msg *shared.KafkaMessage)

MarkMessage marks a message for commit.

func (*Consumer) MarkMessages

func (c *Consumer) MarkMessages(msgs []*shared.KafkaMessage)

MarkMessages marks multiple messages for commit.

func (*Consumer) Start

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

Start runs the Consumer.

type GroupHandler

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

func (*GroupHandler) Cleanup

func (c *GroupHandler) Cleanup(session sarama.ConsumerGroupSession) error

func (*GroupHandler) ConsumeClaim

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

func (*GroupHandler) Setup

func (c *GroupHandler) Setup(_ sarama.ConsumerGroupSession) error

type TopicPartition

type TopicPartition struct {
	Topic     string
	Partition int32
}