Merge pull request #18 from larsnovikov/consistency

Consistency

Author: larsnovikov

.gitignore

@@ -7,7 +7,8 @@ build/
 .DS_Store
 src/.env
 src/vendor
-src/configs/*.json
+src/system/configs/*.json
+src/system/positions/*.txt
 src/plugins/user/*.go
 src/plugins/user/*.so
 src/horgh-replicator

README.md

@@ -16,13 +16,12 @@
 - Configure your my MySQL master as `/mysql/mysql.conf`. 
 Don't forget to set `binlog_do_db=<master_db_name>` and restart MySQL service.
 - Execute `sql/structure.sql` in your MySQL master and slave.
-- Execute `sql/replicator.sql` in your MySQL. It will create database for system values.
 - Start Docker as `make start-dev`
 - Run as `cd src` and `go run main.go listen` in docker container.
 
 ### Testing
 
-- Copy `examples/user.json` and `examples/post.json` to `src/configs`
+- Copy `examples/user.json` and `examples/post.json` to `src/system/configs`
 - Execute `cd src` and `go run main.go load`
 
 ### Add tables to replicator

sql/replicator.sql

@@ -12,12 +12,6 @@ SLAVE_PASS=123456
 SLAVE_TYPE=mysql
 SLAVE_DBNAME=test
 
-REPLICATOR_HOST=localhost
-REPLICATOR_PORT=33061
-REPLICATOR_USER=root
-REPLICATOR_PASS=123456
-REPLICATOR_DBNAME=replicator
-
 ALLOWED_TABLES="user,post"
 CHANNEL_SIZE=99999999
 SLAVE_ID=1

src/.env.dist

@@ -4,8 +4,8 @@ const (
 	PositionPosPrefix  = "position_pos_%s"
 	PositionNamePrefix = "position_name_%s"
 	DBSlave            = "slave"
-	DBReplicator       = "replicator"
 	DBMaster           = "master"
-	ConfigPath         = "configs/%s.json"
+	ConfigPath         = "system/configs/%s.json"
 	PluginPath         = "plugins/%s/handler.so"
+	PositionsPath      = "system/positions/%s.txt"
 )

src/constants/constants.go

@@ -2,7 +2,7 @@ package constants
 
 const (
 	ErrorMysqlCanal          = "Invalid canal"
-	ErrorMysqlPosition       = "Invalid position"
+	ErrorGetPosition         = "Invalid position. Error: \"%s\""
 	ErrorSliceCreation       = "InterfaceSlice() given a non-slice type"
 	ErrorNoColumn            = "There is no column %s in %s.%s"
 	ErrorDBConnect           = "Can't connect to \"%s\""

src/constants/errors.go

@@ -32,7 +32,6 @@ type CredentialsAMQP struct {
 
 var master CredentialsDB
 var slave interface{}
-var replicator CredentialsDB
 
 var tables []string
 var channelSize int
@@ -58,18 +57,6 @@ func MakeCredentials() {
 		os.Getenv("MASTER_DBNAME"),
 	}
 
-	replicationPort, _ := strconv.Atoi(os.Getenv("REPLICATOR_PORT"))
-	replicator = CredentialsDB{
-		Credentials{
-			"mysql",
-		},
-		os.Getenv("REPLICATOR_HOST"),
-		replicationPort,
-		os.Getenv("REPLICATOR_USER"),
-		os.Getenv("REPLICATOR_PASS"),
-		os.Getenv("REPLICATOR_DBNAME"),
-	}
-
 	if os.Getenv("ALLOWED_TABLES") != "" {
 		for _, tableName := range strings.Split(os.Getenv("ALLOWED_TABLES"), ",") {
 			tables = append(tables, strings.TrimSpace(tableName))
@@ -87,8 +74,6 @@ func GetCredentials(storageType string) interface{} {
 		return getMaster()
 	case constants.DBSlave:
 		return getSlave()
-	case constants.DBReplicator:
-		return getReplicator()
 	default:
 		return Credentials{}
 	}
@@ -102,10 +87,6 @@ func getSlave() interface{} {
 	return slave
 }
 
-func getReplicator() CredentialsDB {
-	return replicator
-}
-
 func GetTables() []string {
 	return tables
 }

src/helpers/config.go

@@ -9,14 +9,9 @@ type ConnectionMaster interface {
 	Get(map[string]interface{}) *sql.Rows
 }
 
-type ConnectionReplicator interface {
-	ConnectionMaster
-}
-
 type ConnectionPool struct {
-	Master     ConnectionMaster
-	Slave      Storage
-	Replicator ConnectionReplicator
+	Master ConnectionMaster
+	Slave  Storage
 }
 
 var ConnPool ConnectionPool

src/helpers/connection_pool.go

@@ -12,21 +12,15 @@ func Exec(mode string, params map[string]interface{}) bool {
 	case constants.DBMaster:
 		helpers.ConnPool.Master = mysql.GetConnection(helpers.ConnPool.Master, constants.DBMaster).(helpers.ConnectionMaster)
 		return helpers.ConnPool.Master.Exec(params)
-	case constants.DBReplicator:
-		helpers.ConnPool.Replicator = mysql.GetConnection(helpers.ConnPool.Replicator, constants.DBReplicator).(helpers.ConnectionReplicator)
-		return helpers.ConnPool.Replicator.Exec(params)
 	}
 
 	return false
 }
 
 func Get(mode string, params map[string]interface{}) *sql.Rows {
 	switch mode {
-	case constants.DBMaster:
+	default:
 		helpers.ConnPool.Master = mysql.GetConnection(helpers.ConnPool.Master, constants.DBMaster).(helpers.ConnectionMaster)
 		return helpers.ConnPool.Master.Get(params)
-	default:
-		helpers.ConnPool.Replicator = mysql.GetConnection(helpers.ConnPool.Replicator, constants.DBReplicator).(helpers.ConnectionReplicator)
-		return helpers.ConnPool.Replicator.Get(params)
 	}
 }

src/models/system/connector.go

@@ -1,53 +1,51 @@
 package system
 
 import (
+	"fmt"
+	"github.com/siddontang/go-mysql/mysql"
 	"horgh-replicator/src/constants"
+	"horgh-replicator/src/tools/exit"
+	"io/ioutil"
+	"os"
+	"strconv"
+	"strings"
 )
 
-type replicator struct {
-	Key   string `gorm:"column:param_key"`
-	Value string `gorm:"column:param_value"`
-}
-
-func GetValue(key string) string {
-	query := `SELECT * FROM param_values WHERE param_key=? LIMIT 1;`
-	params := []interface{}{
-		key,
-	}
-
-	res := Get(constants.DBReplicator, map[string]interface{}{
-		"query":  query,
-		"params": params,
-	})
-
-	var row replicator
-	for res.Next() {
-		err := res.Scan(&row.Key, &row.Value)
-		if err != nil {
-			panic(err.Error())
-		}
-	}
-	result := row.Value
+const (
+	PositionMask = "%s:%s"
+)
 
-	defer func() {
-		_ = res.Close()
-	}()
+func SetPosition(hash string, position mysql.Position) error {
+	content := fmt.Sprintf(PositionMask, position.Name, strconv.Itoa(int(position.Pos)))
+	fileName := fmt.Sprintf(constants.PositionsPath, hash)
+	err := ioutil.WriteFile(fileName, []byte(content), 0644)
 
-	return result
+	return err
 }
 
-func SetValue(key string, value string) bool {
-	query := `INSERT INTO param_values(param_key, param_value) VALUES(?, ?) ON DUPLICATE KEY UPDATE param_value=?;`
-	params := []interface{}{
-		key,
-		value,
-		value,
+func GetPosition(hash string) mysql.Position {
+	fileName := fmt.Sprintf(constants.PositionsPath, hash)
+
+	if _, err := os.Stat(fileName); os.IsNotExist(err) {
+		return mysql.Position{}
 	}
 
-	res := Exec(constants.DBReplicator, map[string]interface{}{
-		"query":  query,
-		"params": params,
-	})
+	content, err := ioutil.ReadFile(fileName)
+	if err != nil {
+		exit.Fatal(constants.ErrorGetPosition, err.Error())
+	}
+	data := strings.Split(string(content[:]), ":")
+	if len(data) != 2 {
+		exit.Fatal(constants.ErrorGetPosition, "Can't parse position")
+	}
+	position, err := strconv.Atoi(data[1])
+	if err != nil {
+		exit.Fatal(constants.ErrorGetPosition, err.Error())
+	}
+	pos := mysql.Position{
+		Name: data[0],
+		Pos:  uint32(position),
+	}
 
-	return res
+	return pos
 }