consumer

Documentation

Index

• func TopicFromContext(ctx context.Context) (string, bool)
• type CfgOption
  • func FromNewest() CfgOption
• type Consumer
  • func New(kafkaBrokers []string, consumerGroupID string, topics []string, ...) (*Consumer, error)
  • func (c *Consumer) AddFallback(fn FallbackFunc)
  • func (c *Consumer) AddHandler(messageType string, handler handler.Handler)
  • func (c *Consumer) HealthCheck(_ context.Context) bool
  • func (c *Consumer) Start(ctx context.Context) error
• type FallbackFunc

Functions

func TopicFromContext

func TopicFromContext(ctx context.Context) (string, bool)

Types

type CfgOption

type CfgOption func(config *sarama.Config)

func FromNewest

func FromNewest() CfgOption

type Consumer

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

func New

func New(kafkaBrokers []string, consumerGroupID string, topics []string, cfgOptions ...CfgOption) (*Consumer, error)

func (*Consumer) AddFallback

func (c *Consumer) AddFallback(fn FallbackFunc)

func (*Consumer) AddHandler

func (c *Consumer) AddHandler(messageType string, handler handler.Handler)

func (*Consumer) HealthCheck

func (c *Consumer) HealthCheck(_ context.Context) bool

func (*Consumer) Start

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

type FallbackFunc

type FallbackFunc func(context.Context, []byte) error