messaging

Mar 12, 2018

Changes in this version

+ type Client interface

+ NewConsumer func(topicName, consumerName string, concurrency int) (kafka.Consumer, error)

+ NewProducer func(topicName string) (Producer, error)

+ type ClusterConfig struct

+ Brokers []string

+ type KafkaConfig struct

+ Clusters map[string]ClusterConfig

+ Topics map[string]TopicConfig

+ func (k *KafkaConfig) NewKafkaClient(zLogger *zap.Logger, logger bark.Logger, metricScope tally.Scope) Client

+ type Producer interface

+ Close func() error

+ Publish func(msg *replicator.ReplicationTask) error

+ PublishBatch func(msgs []*replicator.ReplicationTask) error

+ func NewKafkaProducer(topic string, producer sarama.SyncProducer, logger bark.Logger) Producer

+ type TopicConfig struct

+ Cluster string