consumer

Documentation

Index

• type Consumer
  • func NewSaramaConsuemr(config *config.Configuration, consumerGroupID string, topic string, ...) *Consumer
  • func (c *Consumer) Cleanup(_ sarama.ConsumerGroupSession) error
  • func (c *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error
  • func (c *Consumer) Setup(_ sarama.ConsumerGroupSession) error
  • func (c *Consumer) StartConsume(ctx context.Context) (<-chan models.Message, error)

Types

type Consumer

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

func NewSaramaConsuemr

func NewSaramaConsuemr(
	config *config.Configuration,
	consumerGroupID string,
	topic string,
	partitions []int32,
	offset int64,
	messagesLimit int32,
	commit bool,
	follow bool,
	fromTime time.Time,
	toTime time.Time,
) *Consumer

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) Setup

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

func (*Consumer) StartConsume

func (c *Consumer) StartConsume(ctx context.Context) (<-chan models.Message, error)