v1.0.2

Author: wj596

README.md

@@ -118,3 +118,9 @@ server_id=1 # 配置 MySQL replaction 需要定义，不要和 go-mysql-transfer
 
 * 9.22  release
 
+**v1.0.2 release**
+
+* Lua引擎增加数据库操作（dbOps）模块、 http操作(httpOps)模块
+* 修复enum类型出现的乱码问题
+* redis增加Sorted Set数据类型
+

global/config.go

@@ -177,6 +177,8 @@ func NewConfigWithFile(name string) (*Config, error) {
 		if err := checkElsConfig(&c); err != nil {
 			return nil, errors.Trace(err)
 		}
+	default:
+		return nil, errors.Errorf("unsupported target: %s", c.Target)
 	}
 
 	_config = &c
@@ -361,6 +363,10 @@ func checkElsConfig(c *Config) error {
 	return nil
 }
 
+func checkLuaConfig(c *Config) error {
+	return nil
+}
+
 func (c *Config) IsCluster() bool {
 	if !c.IsZk() && !c.IsEtcd() {
 		return false

global/model.go

@@ -18,13 +18,15 @@
 package global
 
 import (
-	"github.com/siddontang/go-mysql/schema"
 	"sync"
+
+	"github.com/siddontang/go-mysql/schema"
 )
 
 type RowRequest struct {
 	RuleKey string
 	Action  string
+	OldRow  []interface{}
 	Row     []interface{}
 }
 
@@ -39,6 +41,8 @@ type RedisRespond struct {
 	Structure string
 	Key       string
 	Field     string
+	Score     float64
+	OldVal    interface{}
 	Val       interface{}
 }
 

global/rule.go

@@ -37,10 +37,11 @@ import (
 )
 
 const (
-	RedisStructureString = "1"
-	RedisStructureHash   = "2"
-	RedisStructureList   = "3"
-	RedisStructureSet    = "4"
+	RedisStructureString    = "String"
+	RedisStructureHash      = "Hash"
+	RedisStructureList      = "List"
+	RedisStructureSet       = "Set"
+	RedisStructureSortedSet = "SortedSet"
 
 	ValEncoderJson     = "json"
 	ValEncoderKVCommas = "kv-commas"
@@ -93,12 +94,14 @@ type Rule struct {
 	RedisHashFieldPrefix string `yaml:"redis_hash_field_prefix"`
 	// 使用哪个列的值作为hash的field，仅redis_structure为hash时起作用
 	RedisHashFieldColumn string `yaml:"redis_hash_field_column"`
-
-	RedisKeyColumnIndex        int
-	RedisKeyColumnIndexs       []int
-	RedisKeyColumnIndexMap     map[string]int
-	RedisHashFieldColumnIndex  int
-	RedisHashFieldColumnIndexs []int
+	// Sorted Set(有序集合)的Score
+	RedisSortedSetScoreColumn      string `yaml:"redis_sorted_set_score_column"`
+	RedisKeyColumnIndex            int
+	RedisKeyColumnIndexs           []int
+	RedisKeyColumnIndexMap         map[string]int
+	RedisHashFieldColumnIndex      int
+	RedisHashFieldColumnIndexs     []int
+	RedisSortedSetScoreColumnIndex int
 
 	// ------------------- ROCKETMQ -----------------
 	RocketmqTopic string `yaml:"rocketmq_topic"` //rocketmq topic名称，可以为空，为空时使用表名称
@@ -191,21 +194,6 @@ func RuleInsList() []*Rule {
 	return list
 }
 
-func StructureName(structure string) string {
-	switch structure {
-	case RedisStructureString:
-		return "string"
-	case RedisStructureHash:
-		return "hash"
-	case RedisStructureList:
-		return "list"
-	case RedisStructureSet:
-		return "set"
-	}
-
-	return ""
-}
-
 func (s *Rule) Initialize() error {
 	if err := s.buildPaddingMap(); err != nil {
 		return err
@@ -513,8 +501,21 @@ func (s *Rule) initRedisConfig() error {
 		if s.RedisKeyValue == "" {
 			return errors.New("empty redis_key_value not allowed in rule")
 		}
+	case "SORTEDSET":
+		s.RedisStructure = RedisStructureSortedSet
+		if s.RedisKeyValue == "" {
+			return errors.New("empty redis_key_value not allowed in rule")
+		}
+		if s.RedisSortedSetScoreColumn == "" {
+			return errors.New("empty redis_sorted_set_score_column not allowed in rule")
+		}
+		_, index := s.TableColumn(s.RedisSortedSetScoreColumn)
+		if index < 0 {
+			return errors.New("redis_sorted_set_score_column must be table column")
+		}
+		s.RedisHashFieldColumnIndex = index
 	default:
-		return errors.Errorf(" redis_structure must be string or hash or list or set")
+		return errors.Errorf("redis_structure must be string or hash or list or set")
 	}
 
 	if s.RedisKeyColumn != "" {
@@ -664,9 +665,11 @@ func (s *Rule) PreCompileLuaScript(dataDir string) error {
 			strings.Contains(script, `HSET(`) ||
 			strings.Contains(script, `RPUSH(`) ||
 			strings.Contains(script, `SADD(`) ||
+			strings.Contains(script, `ZADD(`) ||
 			strings.Contains(script, `DEL(`) ||
 			strings.Contains(script, `HDEL(`) ||
 			strings.Contains(script, `LREM(`) ||
+			strings.Contains(script, `ZREM(`) ||
 			strings.Contains(script, `SREM(`)) {
 
 			return errors.New("lua script incorrect format")

go.mod

@@ -16,11 +16,11 @@ require (
 	github.com/onsi/ginkgo v1.14.0 // indirect
 	github.com/pingcap/errors v0.11.4
 	github.com/pingcap/tidb v1.1.0-beta.0.20191115021711-b274eb2079dc
+	github.com/pkg/errors v0.9.1
 	github.com/pquerna/ffjson v0.0.0-20190930134022-aa0246cd15f7
 	github.com/prometheus/client_golang v1.0.0
 	github.com/samuel/go-zookeeper v0.0.0-20200724154423-2164a8ac840e
 	github.com/satori/go.uuid v1.2.0
-	github.com/shopspring/decimal v0.0.0-20180709203117-cd690d0c9e24
 	github.com/siddontang/go-log v0.0.0-20180807004314-8d05993dda07
 	github.com/siddontang/go-mysql v1.1.0
 	github.com/streadway/amqp v1.0.0
@@ -31,7 +31,6 @@ require (
 	go.mongodb.org/mongo-driver v1.4.0
 	go.uber.org/atomic v1.6.0
 	go.uber.org/zap v1.15.0
-	golang.org/x/tools v0.0.0-20191107010934-f79515f33823
 	gopkg.in/natefinch/lumberjack.v2 v2.0.0
 	gopkg.in/yaml.v2 v2.3.0
 )

main.go

@@ -64,9 +64,8 @@ func main() {
 	n := runtime.GOMAXPROCS(runtime.NumCPU())
 	log.Println(fmt.Sprintf("GOMAXPROCS :%d", n))
 
-
-	//cfgPath  = "D:\\transfer\\app_elasticsearch.yml"
-	//stockFlag   =true
+	cfgPath = "D:\\transfer\\app_redis.yml"
+	//stockFlag = true
 
 	err := service.InitApplication(cfgPath)
 	if err != nil {

service/endpoint/endpoint.go

@@ -29,6 +29,7 @@ import (
 	"github.com/vmihailenco/msgpack"
 
 	"go-mysql-transfer/global"
+	"go-mysql-transfer/service/luaengine"
 	"go-mysql-transfer/storage"
 	"go-mysql-transfer/util/logutil"
 	"go-mysql-transfer/util/stringutil"
@@ -42,7 +43,9 @@ type Endpoint interface {
 	Close()
 }
 
-func NewEndpoint(c *global.Config) Endpoint {
+func NewEndpoint(c *global.Config, ds *canal.Canal) Endpoint {
+	luaengine.InitActuator(ds)
+
 	if c.IsRedis() {
 		return newRedisEndpoint(c)
 	}
@@ -76,23 +79,29 @@ func NewEndpoint(c *global.Config) Endpoint {
 }
 
 func convertColumnData(value interface{}, col *schema.TableColumn, rule *global.Rule) interface{} {
+	if value == nil {
+		return nil
+	}
+
 	switch col.Type {
 	case schema.TYPE_ENUM:
 		switch value := value.(type) {
 		case int64:
-			// for binlog, ENUM may be int64, but for dump, enum is string
 			eNum := value - 1
 			if eNum < 0 || eNum >= int64(len(col.EnumValues)) {
 				// we insert invalid enum value before, so return empty
 				logutil.Warnf("invalid binlog enum index %d, for enum %v", eNum, col.EnumValues)
 				return ""
 			}
 			return col.EnumValues[eNum]
+		case string:
+			return value
+		case []byte:
+			return string(value)
 		}
 	case schema.TYPE_SET:
 		switch value := value.(type) {
 		case int64:
-			// for binlog, SET may be int64, but for dump, SET is string
 			bitmask := value
 			sets := make([]string, 0, len(col.SetValues))
 			for i, s := range col.SetValues {
@@ -105,12 +114,9 @@ func convertColumnData(value interface{}, col *schema.TableColumn, rule *global.
 	case schema.TYPE_BIT:
 		switch value := value.(type) {
 		case string:
-			// for binlog, BIT is int64, but for dump, BIT is string
-			// for dump 0x01 is for 1, \0 is for 0
 			if value == "\x01" {
 				return int64(1)
 			}
-
 			return int64(0)
 		}
 	case schema.TYPE_STRING:
@@ -236,6 +242,28 @@ func keyValueMap(re *global.RowRequest, rule *global.Rule, primitive bool) map[s
 	return kv
 }
 
+func oldKeyValueMap(request *global.RowRequest, rule *global.Rule, primitive bool) map[string]interface{} {
+	kv := make(map[string]interface{}, len(rule.PaddingMap))
+	if primitive {
+		for _, padding := range rule.PaddingMap {
+			kv[padding.ColumnName] = convertColumnData(request.OldRow[padding.ColumnIndex], padding.ColumnMetadata, rule)
+		}
+		return kv
+	}
+
+	for _, padding := range rule.PaddingMap {
+		kv[padding.WrapName] = convertColumnData(request.OldRow[padding.ColumnIndex], padding.ColumnMetadata, rule)
+	}
+
+	if rule.DefaultColumnValueConfig != "" {
+		for k, v := range rule.DefaultColumnValueMap {
+			kv[rule.WrapName(k)] = v
+		}
+	}
+
+	return kv
+}
+
 func primaryKey(re *global.RowRequest, rule *global.Rule) interface{} {
 	if rule.IsCompositeKey { // 组合ID
 		var key string