server

Documentation

Overview

Copyright [2018] [jc3wish]

Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Copyright [2018] [jc3wish]

Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Copyright [2018] [jc3wish]

Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Copyright [2018] [jc3wish]

Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Copyright [2018] [jc3wish]

Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Copyright [2018] [jc3wish]

Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Index

• Constants
• Variables
• func AddNewDB(Name string, ConnectUri string, binlogFileName string, binlogPostion uint32, ...) *db
• func AddTable(db string, schema string, tableName string, channelId int) error
• func AddTableToServer(db string, schemaName string, tableName string, ToServerInfo ToServer) error
• func DelChannel(name string, channelID int) bool
• func DelDB(Name string) bool
• func DelTable(db string, schema string, tableName string) error
• func GetDBObj(Name string) *db
• func GetListDb() map[string]DbListStruct
• func NewConsumeChannel(c *Channel) *consume_channel_obj
• func NewDb(Name string, ConnectUri string, binlogFileName string, binlogPostion uint32, ...) *db
• func Recovery(content *json.RawMessage)
• func SaveDBInfoToFileData() interface{}
• func StopAllChannel()
• type Channel
  • func GetChannel(name string, channelID int) *Channel
  • func NewChannel(MaxThreadNum int, Name string, db *db) *Channel
  • func (Channel *Channel) AddWaitError(WaitErr error, WaitData interface{}) int
  • func (Channel *Channel) Close()
  • func (Channel *Channel) DealWaitError(id int) bool
  • func (Channel *Channel) DelWaitError(id int) bool
  • func (Channel *Channel) GetChannel() chan mysql.EventReslut
  • func (This *Channel) GetChannelMaxThreadNum() int
  • func (Channel *Channel) GetWaitErrorDeal(id int) int
  • func (This *Channel) SetChannelMaxThreadNum(n int)
  • func (Channel *Channel) SetFlowCountChan(flowChan chan *count.FlowCount)
  • func (Channel *Channel) Start() chan mysql.EventReslut
  • func (Channel *Channel) Stop()
• type ChannelErr
• type DbListStruct
• type Table
• type ToServer

Constants

const RegularxEpression = `\{\$([a-zA-Z0-9\-\_]+)\}`

Variables

var DbList map[string]*db

var DbLock sync.Mutex

Functions

func AddNewDB

func AddNewDB(Name string, ConnectUri string, binlogFileName string, binlogPostion uint32, serverId uint32, maxFileName string, maxPosition uint32) *db

func AddTable

func AddTable(db string, schema string, tableName string, channelId int) error

func AddTableToServer

func AddTableToServer(db string, schemaName string, tableName string, ToServerInfo ToServer) error

func DelChannel

func DelChannel(name string, channelID int) bool

func DelDB

func DelDB(Name string) bool

func DelTable

func DelTable(db string, schema string, tableName string) error

func GetDBObj

func GetDBObj(Name string) *db

func GetListDb

func GetListDb() map[string]DbListStruct

func NewConsumeChannel

func NewConsumeChannel(c *Channel) *consume_channel_obj

func NewDb

func NewDb(Name string, ConnectUri string, binlogFileName string, binlogPostion uint32, serverId uint32, maxFileName string, maxPosition uint32) *db

func Recovery

func Recovery(content *json.RawMessage)

func SaveDBInfoToFileData

func SaveDBInfoToFileData() interface{}

func StopAllChannel

func StopAllChannel()

Types

type Channel

type Channel struct {
	sync.Mutex
	Name string

	MaxThreadNum     int // 消费通道的最大线程数
	CurrentThreadNum int
	Status           string //stop ,starting,running,wait

	Errs map[int]*ChannelErr
	// contains filtered or unexported fields
}

func GetChannel

func GetChannel(name string, channelID int) *Channel

func NewChannel

func NewChannel(MaxThreadNum int, Name string, db *db) *Channel

func (*Channel) AddWaitError

func (Channel *Channel) AddWaitError(WaitErr error, WaitData interface{}) int

func (*Channel) Close

func (Channel *Channel) Close()

func (*Channel) DealWaitError

func (Channel *Channel) DealWaitError(id int) bool

func (*Channel) DelWaitError

func (Channel *Channel) DelWaitError(id int) bool

func (*Channel) GetChannel

func (Channel *Channel) GetChannel() chan mysql.EventReslut

func (*Channel) GetChannelMaxThreadNum

func (This *Channel) GetChannelMaxThreadNum() int

func (*Channel) GetWaitErrorDeal

func (Channel *Channel) GetWaitErrorDeal(id int) int

func (*Channel) SetChannelMaxThreadNum

func (This *Channel) SetChannelMaxThreadNum(n int)

func (*Channel) SetFlowCountChan

func (Channel *Channel) SetFlowCountChan(flowChan chan *count.FlowCount)

func (*Channel) Start

func (Channel *Channel) Start() chan mysql.EventReslut

func (*Channel) Stop

func (Channel *Channel) Stop()

type ChannelErr

type ChannelErr struct {
	WaitErr  error
	WaitData interface{}
	WaitDeal int // 0不处理,1错过
}

type DbListStruct

type DbListStruct struct {
	Name                  string
	ConnectUri            string
	ConnStatus            string //close,stop,starting,running
	ConnErr               string
	ChannelCount          int
	LastChannelID         int
	TableCount            int
	BinlogDumpFileName    string
	BinlogDumpPosition    uint32
	MaxBinlogDumpFileName string
	MaxBinlogDumpPosition uint32
	ReplicateDoDb         map[string]uint8
	ServerId              uint32
}

type Table

type Table struct {
	sync.Mutex
	Name         string
	ChannelKey   int
	ToServerList []ToServer
}

type ToServer

type ToServer struct {
	MustBeSuccess bool
	Type          string
	DataType      string //string , json
	AddEventType  bool
	AddSchemaName bool
	AddTableName  bool
	KeyConfig     string
	ValueConfig   string
	FieldList     []string
	ToServerKey   string
	Expir         int
}