kafkautils

Documentation

Index

• Variables
• func CreateErrorLogger(l *zap.SugaredLogger) kafka.Logger
• func CreateLogger(l *zap.SugaredLogger) kafka.Logger
• func ProtoTypeFromHeaders(headers []kafka.Header) (string, error)
• type ConsumerHandler
  • func NewConsumerHandler(logger *zap.SugaredLogger, reader *kafka.Reader) ConsumerHandler

Variables

var ErrNoProtoTypeHeader = fmt.Errorf("no proto type found in message headers")

Functions

func CreateErrorLogger

func CreateErrorLogger(l *zap.SugaredLogger) kafka.Logger

func CreateLogger

func CreateLogger(l *zap.SugaredLogger) kafka.Logger

func ProtoTypeFromHeaders

func ProtoTypeFromHeaders(headers []kafka.Header) (string, error)

Types

type ConsumerHandler

type ConsumerHandler interface {
	// Run starts the consumer handler
	// It will block until the context is cancelled
	Run(ctx context.Context)

	RegisterHandler(protoType proto.Message, consumer func(context.Context, *kafka.Message, proto.Message))
}

func NewConsumerHandler

func NewConsumerHandler(logger *zap.SugaredLogger,
	reader *kafka.Reader) ConsumerHandler