kafka

Documentation

Index

• type KafkaReader
  • func NewKafkaReader(config KafkaReaderConfig) (*KafkaReader, error)
  • func (kr *KafkaReader) Close() error
  • func (kr *KafkaReader) Read() ([]byte, error)
• type KafkaReaderConfig
• type KafkaWritter
  • func NewKafkaWritter(config KafkaWritterConfig) (*KafkaWritter, error)
  • func (kw *KafkaWritter) Close() error
  • func (kw *KafkaWritter) Write(msgs []Message)
• type KafkaWritterConfig
• type Message

Types

type KafkaReader

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

func NewKafkaReader

func NewKafkaReader(config KafkaReaderConfig) (*KafkaReader, error)

func (*KafkaReader) Close

func (kr *KafkaReader) Close() error

func (*KafkaReader) Read

func (kr *KafkaReader) Read() ([]byte, error)

type KafkaReaderConfig

type KafkaReaderConfig struct {
	Brokers       []string
	GroupId       string
	Topic         string
	QueueCapacity int
	MinBytes      int
	MaxBytes      int
}

type KafkaWritter

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

func NewKafkaWritter

func NewKafkaWritter(config KafkaWritterConfig) (*KafkaWritter, error)

func (*KafkaWritter) Close

func (kw *KafkaWritter) Close() error

func (*KafkaWritter) Write

func (kw *KafkaWritter) Write(msgs []Message)

type KafkaWritterConfig

type KafkaWritterConfig struct {
	Brokers      []string
	Topic        string
	MaxAttempts  int
	BatchSize    int
	BatchTimeout time.Duration
	WriteTimeout time.Duration
	LogFileName  string
}

type Message

type Message []byte