kafka

Nov 17, 2022

Changes in this version

type FTMessage

+ Topic string

Jul 5, 2022

Changes in this version

+ const LagTechnicalSummary

May 17, 2022

Changes in this version

type ConsumerConfig

+ DisableMonitoringConnectionReset bool

Mar 30, 2022

Changes in this version

+ var ErrConsumerNotConnected = fmt.Errorf("consumer is not connected to Kafka")

+ var ErrMonitorNotConnected = fmt.Errorf("consumer monitor is not connected to Kafka")

+ var ErrProducerNotConnected = fmt.Errorf("producer is not connected to Kafka")

+ func DefaultConsumerOptions() *sarama.Config

+ func DefaultProducerOptions() *sarama.Config

+ type Consumer struct

+ func NewConsumer(config ConsumerConfig, topics []*Topic, log *logger.UPPLogger) *Consumer

+ func (c *Consumer) Close() error

+ func (c *Consumer) ConnectivityCheck() error

+ func (c *Consumer) MonitorCheck() error

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

+ type ConsumerConfig struct

+ BrokersConnectionString string

+ ConnectionRetryInterval time.Duration

+ ConsumerGroup string

+ OffsetFetchInterval time.Duration

+ OffsetFetchMaxFailureCount int

+ Options *sarama.Config

+ 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

+ ConnectionRetryInterval time.Duration

+ Options *sarama.Config

+ Topic string

+ type Topic struct

+ Name string

+ func NewTopic(name string, opts ...TopicOption) *Topic

+ type TopicOption func(topic *Topic)

+ func WithLagTolerance(tolerance int64) TopicOption