client

Documentation

Index

• func ByteSliceToHexString(data []byte) string
• func Scramble411(data *[]byte, seed *[]byte) []byte
• type CanalClusterNode
  • func NewCanalClusterNode(destination string, zkServer []string, timeout time.Duration) (canalNode *CanalClusterNode, err error)
  • func (canalNode *CanalClusterNode) GetNode() (addr string, port int, err error)
  • func (canalNode *CanalClusterNode) InitClusters(addressList []string)
• type CanalConnector
• type ClusterCanalConnector
  • func NewClusterCanalConnector(canalNode *CanalClusterNode, username string, password string, ...) (*ClusterCanalConnector, error)
  • func (cc *ClusterCanalConnector) Ack(batchId int64) error
  • func (cc *ClusterCanalConnector) Connect() error
  • func (cc *ClusterCanalConnector) DisConnection() error
  • func (cc *ClusterCanalConnector) Get(batchSize int32, timeOut *int64, units *int32) (msg *pb.Message, err error)
  • func (cc *ClusterCanalConnector) GetWithOutAck(batchSize int32, timeOut *int64, units *int32) (msg *pb.Message, err error)
  • func (cc *ClusterCanalConnector) RollBack(batchId int64) error
  • func (cc *ClusterCanalConnector) Subscribe(filter string) error
  • func (cc *ClusterCanalConnector) UnSubscribe() error
• type ServerRunningData
• type SimpleCanalConnector
  • func NewSimpleCanalConnector(address string, port int, username string, password string, destination string, ...) *SimpleCanalConnector
  • func (c *SimpleCanalConnector) Ack(batchId int64) error
  • func (c *SimpleCanalConnector) Connect() error
  • func (c *SimpleCanalConnector) DisConnection() error
  • func (c *SimpleCanalConnector) Get(batchSize int32, timeOut *int64, units *int32) (*pb.Message, error)
  • func (c *SimpleCanalConnector) GetWithOutAck(batchSize int32, timeOut *int64, units *int32) (*pb.Message, error)
  • func (c *SimpleCanalConnector) RollBack(batchId int64) error
  • func (c *SimpleCanalConnector) Subscribe(filter string) error
  • func (c *SimpleCanalConnector) UnSubscribe() error
  • func (c *SimpleCanalConnector) WriteWithHeader(body []byte)

Functions

func ByteSliceToHexString

func ByteSliceToHexString(data []byte) string

func Scramble411

func Scramble411(data *[]byte, seed *[]byte) []byte

Types

type CanalClusterNode

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

func NewCanalClusterNode

func NewCanalClusterNode(destination string, zkServer []string, timeout time.Duration) (canalNode *CanalClusterNode, err error)

func (*CanalClusterNode) GetNode

func (canalNode *CanalClusterNode) GetNode() (addr string, port int, err error)

func (*CanalClusterNode) InitClusters

func (canalNode *CanalClusterNode) InitClusters(addressList []string)

type CanalConnector

type CanalConnector interface {
	Connect() error
	DisConnection() error
	Subscribe(filter string) error
	UnSubscribe() error
	Get(batchSize int32, timeOut *int64, units *int32) (msg *pb.Message, err error)
	GetWithOutAck(batchSize int32, timeOut *int64, units *int32) (msg *pb.Message, err error)
	Ack(batchId int64) error
	RollBack(batchId int64) error
}

type ClusterCanalConnector

type ClusterCanalConnector struct {
	RetryTimes int32

	Path string
	// contains filtered or unexported fields
}

func NewClusterCanalConnector

func NewClusterCanalConnector(canalNode *CanalClusterNode, username string, password string, destination string,
	soTimeOut int32, idleTimeOut int32) (*ClusterCanalConnector, error)

func (*ClusterCanalConnector) Ack

func (cc *ClusterCanalConnector) Ack(batchId int64) error

func (*ClusterCanalConnector) Connect

func (cc *ClusterCanalConnector) Connect() error

func (*ClusterCanalConnector) DisConnection

func (cc *ClusterCanalConnector) DisConnection() error

func (*ClusterCanalConnector) Get

func (cc *ClusterCanalConnector) Get(batchSize int32, timeOut *int64, units *int32) (msg *pb.Message, err error)

func (*ClusterCanalConnector) GetWithOutAck

func (cc *ClusterCanalConnector) GetWithOutAck(batchSize int32, timeOut *int64, units *int32) (msg *pb.Message, err error)

func (*ClusterCanalConnector) RollBack

func (cc *ClusterCanalConnector) RollBack(batchId int64) error

func (*ClusterCanalConnector) Subscribe

func (cc *ClusterCanalConnector) Subscribe(filter string) error

func (*ClusterCanalConnector) UnSubscribe

func (cc *ClusterCanalConnector) UnSubscribe() error

type ServerRunningData

type ServerRunningData struct {
	Cid     int64
	Address string
	Active  bool
}

type SimpleCanalConnector

type SimpleCanalConnector struct {
	Address           string
	Port              int
	UserName          string
	PassWord          string
	SoTime            int32
	IdleTimeOut       int32
	ClientIdentity    pb.ClientIdentity
	Connected         bool
	Running           bool
	Filter            string
	RollbackOnConnect bool
	LazyParseEntry    bool
	// contains filtered or unexported fields
}

func NewSimpleCanalConnector

func NewSimpleCanalConnector(address string, port int, username string, password string, destination string, soTimeOut int32, idleTimeOut int32) *SimpleCanalConnector

NewSimpleCanalConnector 创建SimpleCanalConnector实例

func (*SimpleCanalConnector) Ack

func (c *SimpleCanalConnector) Ack(batchId int64) error

Ack Ack Canal-server的数据（就是昨晚某些逻辑操作后删除canal-server端的数据）

func (*SimpleCanalConnector) Connect

func (c *SimpleCanalConnector) Connect() error

Connect 连接Canal-server

func (*SimpleCanalConnector) DisConnection

func (c *SimpleCanalConnector) DisConnection() error

DisConnection 关闭连接

func (*SimpleCanalConnector) Get

func (c *SimpleCanalConnector) Get(batchSize int32, timeOut *int64, units *int32) (*pb.Message, error)

Get 获取数据并且Ack数据

func (*SimpleCanalConnector) GetWithOutAck

func (c *SimpleCanalConnector) GetWithOutAck(batchSize int32, timeOut *int64, units *int32) (*pb.Message, error)

GetWithOutAck 获取数据不Ack

func (*SimpleCanalConnector) RollBack

func (c *SimpleCanalConnector) RollBack(batchId int64) error

RollBack 回滚操作

func (*SimpleCanalConnector) Subscribe

func (c *SimpleCanalConnector) Subscribe(filter string) error

Subscribe 订阅

func (*SimpleCanalConnector) UnSubscribe

func (c *SimpleCanalConnector) UnSubscribe() error

UnSubscribe 取消订阅

func (*SimpleCanalConnector) WriteWithHeader

func (c *SimpleCanalConnector) WriteWithHeader(body []byte)

WriteWithHeader 写数据包的header+body