reader

Documentation

Index

• type Config
• type KafkaSeeker
  • func NewKafkaSeeker(addr []string, config *sarama.Config) (*KafkaSeeker, error)
  • func (ks *KafkaSeeker) Close()
  • func (ks *KafkaSeeker) Seek(topic string, ts int64, partitions []int32) (offsets []int64, err error)
• type Message
• type Reader
  • func NewReader(cfg *Config) (r *Reader, err error)
  • func (r *Reader) Close()
  • func (r *Reader) Messages() (msgs <-chan *Message)
  • func (r *Reader) Run()

Types

type Config

type Config struct {
	KafkaAddr []string
	// the CommitTs of binlog return by reader will bigger than the config CommitTs
	CommitTS  int64
	Offset    int64 // start at kafka offset
	ClusterID string
	Name      string
}

Config for Reader

type KafkaSeeker

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

KafkaSeeker seeks offset in kafka topics by given condition

func NewKafkaSeeker

func NewKafkaSeeker(addr []string, config *sarama.Config) (*KafkaSeeker, error)

NewKafkaSeeker creates an instance of KafkaSeeker

func (*KafkaSeeker) Close

func (ks *KafkaSeeker) Close()

Close releases resources of KafkaSeeker

func (*KafkaSeeker) Seek

func (ks *KafkaSeeker) Seek(topic string, ts int64, partitions []int32) (offsets []int64, err error)

Seek seeks the first offset which binlog CommitTs bigger than ts

type Message

type Message struct {
	Binlog *pb.Binlog
	Offset int64 // kafka offset
}

Message read from reader

type Reader

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

Reader to read binlog from kafka

func NewReader

func NewReader(cfg *Config) (r *Reader, err error)

NewReader creates an instance of Reader

func (*Reader) Close

func (r *Reader) Close()

Close shuts down the reader

func (*Reader) Messages

func (r *Reader) Messages() (msgs <-chan *Message)

Messages returns a chan that contains unread buffered message

func (*Reader) Run

func (r *Reader) Run()

Run start consume msg