kafka

Documentation

Index

• type BinlogReader
  • func NewBinlogReader(tgt *client.Conn, reader *stdkafka.Reader, logger *zap.Logger) *BinlogReader
  • func (b *BinlogReader) LoadColumnsForTable(fullTable string) ([]string, error)
  • func (b *BinlogReader) Run()
• type BinlogWriter
  • func NewBinlogWriter(writer *stdkafka.Writer, serverId uint32, host string, port uint16, ...) *BinlogWriter
  • func (b *BinlogWriter) Run()
• type CommandSQL
• type Messages
  • func NewMessages(writer *stdkafka.Writer) *Messages
  • func (m *Messages) Push(ctx context.Context, cmd CommandSQL, query []byte)
• type MessagesLogger
  • func NewMessagesLogger(msg Msg, log *zap.Logger) *MessagesLogger
  • func (ml *MessagesLogger) Push(ctx context.Context, cmd CommandSQL, query []byte)
• type Msg

Types

type BinlogReader

type BinlogReader struct {
	Target *client.Conn
	// contains filtered or unexported fields
}

func NewBinlogReader

func NewBinlogReader(tgt *client.Conn, reader *stdkafka.Reader, logger *zap.Logger) *BinlogReader

func (*BinlogReader) LoadColumnsForTable

func (b *BinlogReader) LoadColumnsForTable(fullTable string) ([]string, error)

func (*BinlogReader) Run

func (b *BinlogReader) Run()

type BinlogWriter

type BinlogWriter struct {
	Tables map[string][]config.Table
	// contains filtered or unexported fields
}

func NewBinlogWriter

func NewBinlogWriter(writer *stdkafka.Writer, serverId uint32, host string, port uint16, user, password string, t map[string][]config.Table, gtid, position string, logger *zap.Logger) *BinlogWriter

func (*BinlogWriter) Run

func (b *BinlogWriter) Run()

type CommandSQL

type CommandSQL string

var (
	Transaction CommandSQL = "transaction"
	Commit      CommandSQL = "commit"
	Rollback    CommandSQL = "rollback"
	Insert      CommandSQL = "insert"
	Delete      CommandSQL = "delete"
	Update      CommandSQL = "update"
	Select      CommandSQL = "select"
)

type Messages

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

func NewMessages

func NewMessages(writer *stdkafka.Writer) *Messages

func (*Messages) Push

func (m *Messages) Push(ctx context.Context, cmd CommandSQL, query []byte)

type MessagesLogger

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

func NewMessagesLogger

func NewMessagesLogger(msg Msg, log *zap.Logger) *MessagesLogger

func (*MessagesLogger) Push

func (ml *MessagesLogger) Push(ctx context.Context, cmd CommandSQL, query []byte)

type Msg

type Msg interface {
	Push(ctx context.Context, cmd CommandSQL, query []byte)
}