producer

Dec 15, 2021

Changes in this version

+ func LoadCfgs() (r map[string]*Cfg, err error)

type Cfg

+ func LoadCfg(name string) (r *Cfg, err error)

Nov 26, 2021

Changes in this version

type KafkaProducer

+ func DefaultProducer() (r *KafkaProducer)

Sep 6, 2021

Changes in this version

+ type Partitioner interface

+ Partition func(message *kafka.Message, numPartitions int32) (r int32, err error)

+ func NewHashPartitioner() Partitioner

+ func NewManualPartitioner() Partitioner

+ func NewRandomPartitioner() Partitioner

+ func NewRoundRobinPartitioner() Partitioner

+ type PartitionerConstructor func() Partitioner

Jun 9, 2021

Changes in this version

type Cfg

+ Partitioner string

Apr 19, 2021

Changes in this version

type Cfg

+ Topic string

Apr 16, 2021

Changes in this version

+ var DefaultFailFeedbackFn = func(fail *sarama.ProducerError, msg []byte)

+ var DefaultSucFeedbackFn = func(suc *sarama.ProducerMessage, msg []byte)

+ func CloseDefaultProducer()

+ func CloseProducer(name string)

+ func DefaultProducerPublish(topic string, key string, msg []byte) (err error)

+ func Publish(name string, topic string, key string, msg []byte) (err error)

+ func SetFailFeedback(name string, fn func(*sarama.ProducerError, []byte)) (err error)

+ func SetSucFeedback(name string, fn func(*sarama.ProducerMessage, []byte)) (err error)

+ type Cfg struct

+ Acks int16

+ Addrs []string

+ ReturnErrors bool

+ ReturnFeedbackNum int

+ ReturnSuccesses bool

+ type KafkaProducer struct

+ func NewKafkaProducer(cfg *Cfg) (r *KafkaProducer, err error)

+ func Producer(name string) (r *KafkaProducer)

+ func SafeProducer(name string) (r *KafkaProducer, err error)

+ func (this *KafkaProducer) Close()

+ func (this *KafkaProducer) Publish(topic string, key string, msg []byte)

+ func (this *KafkaProducer) SetFailFeedback(fn func(*sarama.ProducerError, []byte))

+ func (this *KafkaProducer) SetSucFeedback(fn func(*sarama.ProducerMessage, []byte))