broker

Documentation

Index

• type BinlogBrokerConfig
• type BinlogKafkaBroker
  • func New(cfg *BinlogBrokerConfig) (*BinlogKafkaBroker, error)
  • func (b *BinlogKafkaBroker) Consume(fn func(*types.BinlogEvent) error) error
  • func (b *BinlogKafkaBroker) Marshal(event *river.EventData) ([]byte, error)
  • func (b *BinlogKafkaBroker) OnAlert(msg *river.StatusMsg) error
  • func (b *BinlogKafkaBroker) OnClose(r *river.River)
  • func (b *BinlogKafkaBroker) Pipe(river *river.River, from river.From) error
  • func (b *BinlogKafkaBroker) String() string
• type TxKafkaBroker
  • func NewTxKafkaBroker(addrs []string, txInfoTopic string) (*TxKafkaBroker, error)
  • func (k *TxKafkaBroker) Consume(fn func(info *types.TxInfo) error) error
  • func (k *TxKafkaBroker) PushTx(txInfo *types.TxInfo) error

Types

type BinlogBrokerConfig

type BinlogBrokerConfig struct {
	KafkaConfig *kafka.Config
	Tables      []string
}

type BinlogKafkaBroker

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

func New

func New(cfg *BinlogBrokerConfig) (*BinlogKafkaBroker, error)

func (*BinlogKafkaBroker) Consume

func (b *BinlogKafkaBroker) Consume(fn func(*types.BinlogEvent) error) error

Consume 消费kafka中的数据

func (*BinlogKafkaBroker) Marshal

func (b *BinlogKafkaBroker) Marshal(event *river.EventData) ([]byte, error)

func (*BinlogKafkaBroker) OnAlert

func (b *BinlogKafkaBroker) OnAlert(msg *river.StatusMsg) error

func (*BinlogKafkaBroker) OnClose

func (b *BinlogKafkaBroker) OnClose(r *river.River)

func (*BinlogKafkaBroker) Pipe

func (b *BinlogKafkaBroker) Pipe(river *river.River, from river.From) error

Pipe 将river中的数据流向kafka

func (*BinlogKafkaBroker) String

func (b *BinlogKafkaBroker) String() string

type TxKafkaBroker

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

func NewTxKafkaBroker

func NewTxKafkaBroker(addrs []string, txInfoTopic string) (*TxKafkaBroker, error)

func (*TxKafkaBroker) Consume

func (k *TxKafkaBroker) Consume(fn func(info *types.TxInfo) error) error

func (*TxKafkaBroker) PushTx

func (k *TxKafkaBroker) PushTx(txInfo *types.TxInfo) error