iokafka

Documentation

Index

• type KafkaMessage
  • func (m KafkaMessage) ToMessage() Message
• type Message
  • func (m Message) ToKafkaMessage() kafka.Message
• type Scanner
  • func NewScanner(config ScannerConfig) Scanner
  • func (s *Scanner) Err() error
  • func (s *Scanner) Message() Message
  • func (s *Scanner) Scan() bool
• type ScannerConfig
• type Writer
  • func NewWriter(brokerUrl string, topic string) *Writer
  • func (w *Writer) Close() (err error)
  • func (w *Writer) Write(msg Message) (err error)

Types

type KafkaMessage

type KafkaMessage kafka.Message

func (KafkaMessage) ToMessage

func (m KafkaMessage) ToMessage() Message

type Message

type Message struct {
	Topic     string
	Partition int
	Offset    int64
	Key       []byte
	Value     []byte
}

func (Message) ToKafkaMessage

func (m Message) ToKafkaMessage() kafka.Message

type Scanner

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

func NewScanner

func NewScanner(config ScannerConfig) Scanner

func (*Scanner) Err

func (s *Scanner) Err() error

func (*Scanner) Message

func (s *Scanner) Message() Message

func (*Scanner) Scan

func (s *Scanner) Scan() bool

type ScannerConfig

type ScannerConfig struct {
	Brokers       []string
	GroupID       string
	Partition     int
	Topic         string
	AttemtsOnFail int
	FailTimeout   int
	StartOffset   int64
}

type Writer

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

func NewWriter

func NewWriter(brokerUrl string, topic string) *Writer

func (*Writer) Close

func (w *Writer) Close() (err error)

func (*Writer) Write

func (w *Writer) Write(msg Message) (err error)