consumer

Documentation

Index

• type ConfluentConsumer
  • func NewConfluentConsumer(kafkaURLs []string, topic, groupID string) (*ConfluentConsumer, error)
  • func (confluent ConfluentConsumer) Close()
  • func (confluent ConfluentConsumer) Subscribe(f func(message *message.Message) error)
• type Consumer
• type GroupHandler
  • func (h GroupHandler) Cleanup(session sarama.ConsumerGroupSession) error
  • func (h GroupHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error
  • func (h GroupHandler) Setup(session sarama.ConsumerGroupSession) error
• type SaramaConsumer
  • func NewSaramaConsumer(kafkaURLs []string, topic, groupID string) (*SaramaConsumer, error)
  • func (sarama SaramaConsumer) Close()
  • func (sarama SaramaConsumer) Subscribe(f func(message *message.Message) error)
• type SegmentioConsumer
  • func NewSegmentioConsumer(kafkaURLs []string, topic, groupID string) (*SegmentioConsumer, error)
  • func (segmentio SegmentioConsumer) Close()
  • func (segmentio SegmentioConsumer) Subscribe(f func(message *message.Message) error)

Types

type ConfluentConsumer

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

func NewConfluentConsumer

func NewConfluentConsumer(kafkaURLs []string, topic, groupID string) (*ConfluentConsumer, error)

NewConfluentConsumer create new Confluent Consumer.

func (ConfluentConsumer) Close

func (confluent ConfluentConsumer) Close()

func (ConfluentConsumer) Subscribe

func (confluent ConfluentConsumer) Subscribe(f func(message *message.Message) error)

type Consumer

type Consumer interface {
	Subscribe(f func(message *message.Message) error)
	Close()
}

Consumer kafka.

type GroupHandler

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

ConsumerGroupHandler represents the sarama consumer group.

func (GroupHandler) Cleanup

func (h GroupHandler) Cleanup(session sarama.ConsumerGroupSession) error

func (GroupHandler) ConsumeClaim

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

ConsumeClaim must start a consumer loop of ConsumerGroupClaim's Messages(), here is supposed to be what you want to do with the message. In this example the message will be logged with the topic name, partition and message value.

func (GroupHandler) Setup

func (h GroupHandler) Setup(session sarama.ConsumerGroupSession) error

type SaramaConsumer

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

func NewSaramaConsumer

func NewSaramaConsumer(kafkaURLs []string, topic, groupID string) (*SaramaConsumer, error)

NewSaramaConsumer Create new Sarama Consumer.

func (SaramaConsumer) Close

func (sarama SaramaConsumer) Close()

func (SaramaConsumer) Subscribe

func (sarama SaramaConsumer) Subscribe(f func(message *message.Message) error)

type SegmentioConsumer

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

func NewSegmentioConsumer

func NewSegmentioConsumer(kafkaURLs []string, topic, groupID string) (*SegmentioConsumer, error)

NewSegmentioConsumer create new Segmentio Consumer.

func (SegmentioConsumer) Close

func (segmentio SegmentioConsumer) Close()

func (SegmentioConsumer) Subscribe

func (segmentio SegmentioConsumer) Subscribe(f func(message *message.Message) error)