client

Documentation

Index

• func WriteWithHeader(body []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()
  • 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

Functions

func WriteWithHeader

func WriteWithHeader(body []byte)

WriteWithHeader 写数据包的header+body

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()
	Disconnect()
	CheckValid()
	Subscribe()
	UnSubscribe()
	Get(arg ...interface{}) pb.Message
	GetWithoutAck(arg ...interface{}) pb.Message
	Ack(batchId int64)
	Rollback(arg ...interface{})
	StopRunning()
}

type ClusterCanalConnector

type ClusterCanalConnector struct {
	RetryTimes int32
	// 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
}

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()

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 取消订阅