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, ...)

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) (*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.