kafka

Oct 28, 2021

Changes in this version

+ func DefaultConsumerOptions() *sarama.Config

+ func DefaultProducerOptions() *sarama.Config

+ type Consumer struct

+ func NewConsumer(config ConsumerConfig, log *logger.UPPLogger, retryInterval time.Duration) *Consumer

+ func (c *Consumer) Close() error

+ func (c *Consumer) ConnectivityCheck() error

+ func (c *Consumer) StartListening(messageHandler func(message FTMessage))

+ type ConsumerConfig struct

+ BrokersConnectionString string

+ ConsumerGroup string

+ Options *sarama.Config

+ Topics []string

+ type FTMessage struct

+ Body string

+ Headers map[string]string

+ func NewFTMessage(headers map[string]string, body string) FTMessage

+ func (m *FTMessage) Build() string

+ type Producer struct

+ func NewProducer(config ProducerConfig, logger *logger.UPPLogger, ...) *Producer

+ func (p *Producer) Close() error

+ func (p *Producer) ConnectivityCheck() error

+ func (p *Producer) SendMessage(message FTMessage) error

+ type ProducerConfig struct

+ BrokersConnectionString string

+ Options *sarama.Config

+ Topic string