kafka

Dec 19, 2023

Changes in this version

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

+ func InitTopicWithPartitionNumber(bootstrapUrl string, partitionNumber int, topics ...string) (err error)

+ func NewConsumer(ctx context.Context, config configuration.Config, topic string, ...) error

+ func NewConsumerWithFullMessage(ctx context.Context, config configuration.Config, topic string, ...) error

+ type KeySeparationBalancer struct

+ Seperator string

+ SubBalancer kafka.Balancer

+ func (this *KeySeparationBalancer) Balance(msg kafka.Message, partitions ...int) (partition int)

+ type Producer struct

+ func NewProducer(ctx context.Context, config configuration.Config, topic string) (*Producer, error)

+ func NewProducerWithBalancer(ctx context.Context, config configuration.Config, topic string, ...) (*Producer, error)

+ func NewProducerWithKeySeparationBalancer(ctx context.Context, config configuration.Config, topic string) (*Producer, error)

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