kafkaclient

Documentation

Index

• type KafkaClient
  • func NewKafkaClient(broker string, maxRetries int, retryInterval time.Duration, ...) (*KafkaClient, error)
  • func (kc *KafkaClient) Close() error
  • func (kc *KafkaClient) IsReady() bool
  • func (kc *KafkaClient) SendBatch(batch []*sarama.ProducerMessage, topic string) error
  • func (kc *KafkaClient) StartBatchSender(ringBuffer *kafkabuff.RingBuffer, batchSize int, interval time.Duration, ...)
• type Logger
• type StdLogger
  • func (l *StdLogger) Debugf(format string, args ...interface{})
  • func (l *StdLogger) Errorf(format string, args ...interface{})
  • func (l *StdLogger) Infof(format string, args ...interface{})
  • func (l *StdLogger) Warnf(format string, args ...interface{})

Types

type KafkaClient

type KafkaClient struct {
	// contains filtered or unexported fields
}

func NewKafkaClient

func NewKafkaClient(broker string, maxRetries int, retryInterval time.Duration, config *sarama.Config, logger Logger) (*KafkaClient, error)

NewKafkaClient creates a new KafkaClient instance.

func (*KafkaClient) Close

func (kc *KafkaClient) Close() error

Close closes the connection to Kafka.

func (*KafkaClient) IsReady

func (kc *KafkaClient) IsReady() bool

IsReady returns the readiness state of the Kafka client.

func (*KafkaClient) SendBatch

func (kc *KafkaClient) SendBatch(batch []*sarama.ProducerMessage, topic string) error

SendBatch sends a batch of messages to Kafka with retries.

func (*KafkaClient) StartBatchSender

func (kc *KafkaClient) StartBatchSender(ringBuffer *kafkabuff.RingBuffer, batchSize int, interval time.Duration, kafkaTopic string)

StartBatchSender starts sending messages from the ring buffer to Kafka.

type Logger

type Logger interface {
	Debugf(format string, args ...interface{})
	Infof(format string, args ...interface{})
	Warnf(format string, args ...interface{})
	Errorf(format string, args ...interface{})
}

Logger interface

type StdLogger

type StdLogger struct{}

func (*StdLogger) Debugf

func (l *StdLogger) Debugf(format string, args ...interface{})

func (*StdLogger) Errorf

func (l *StdLogger) Errorf(format string, args ...interface{})

func (*StdLogger) Infof

func (l *StdLogger) Infof(format string, args ...interface{})

func (*StdLogger) Warnf

func (l *StdLogger) Warnf(format string, args ...interface{})