kafka

Mar 22, 2019

Changes in this version

type Option

+ func WithAnalytics(on bool) Option

+ func WithAnalyticsRate(rate float64) Option

Sep 13, 2018

Changes in this version

+ type Consumer struct

+ func NewConsumer(conf *kafka.ConfigMap, opts ...Option) (*Consumer, error)

+ func WrapConsumer(c *kafka.Consumer, opts ...Option) *Consumer

+ func (c *Consumer) Close() error

+ func (c *Consumer) Events() chan kafka.Event

+ func (c *Consumer) Poll(timeoutMS int) (event kafka.Event)

+ type MessageCarrier struct

+ func NewMessageCarrier(msg *kafka.Message) MessageCarrier

+ func (c MessageCarrier) ForeachKey(handler func(key, val string) error) error

+ func (c MessageCarrier) Set(key, val string)

+ type Option func(cfg *config)

+ func WithContext(ctx context.Context) Option

+ func WithServiceName(serviceName string) Option

+ type Producer struct

+ func NewProducer(conf *kafka.ConfigMap, opts ...Option) (*Producer, error)

+ func WrapProducer(p *kafka.Producer, opts ...Option) *Producer

+ func (p *Producer) Close()

+ func (p *Producer) Produce(msg *kafka.Message, deliveryChan chan kafka.Event) error

+ func (p *Producer) ProduceChannel() chan *kafka.Message