Documentation
¶
Index ¶
Constants ¶
This section is empty.
Variables ¶
View Source
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 ¶
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
Source Files
Click to show internal directories.
Click to hide internal directories.