kafka

Documentation

Index

• Variables
• func GetBroker(bootstrapUrl string) (brokers []string, err error)
• func InitTopic(bootstrapUrl string, topics ...string) (err error)
• func NewConsumer(ctx context.Context, config configuration.Config, topic string, ...) error
• func NewProducer(ctx context.Context, kafkaUrl string, topic string, debug bool) (interfaces.Producer, error)
• type Producer
  • func (this *Producer) Produce(key string, message []byte) error

Variables

var CommitError = errors.New("unable to commit to kafka")

var FetchError = errors.New("unable to fetch from kafka")

var HandlerError = errors.New("unable to handle kafka message")

Functions

func GetBroker

func GetBroker(bootstrapUrl string) (brokers []string, err error)

func InitTopic

func InitTopic(bootstrapUrl string, topics ...string) (err error)

func NewConsumer

func NewConsumer(ctx context.Context, config configuration.Config, topic string, listener func(delivery []byte) error, errorhandler func(err error) (fatal bool)) error

func NewProducer

func NewProducer(ctx context.Context, kafkaUrl string, topic string, debug bool) (interfaces.Producer, error)

Types

type Producer

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

func (*Producer) Produce

func (this *Producer) Produce(key string, message []byte) error