Merge branch 'branch-v0.2'

Author: GuoYuefei

.gitignore

@@ -1,5 +1,5 @@
 /.idea
-/datatmp
-/tmp
-/data*
+data*
 go.sum
+build/
+!distributed/data

README.md

@@ -68,7 +68,9 @@ LISTEN_ADDRESS=:12345 STORAGE_ROOT=./tmp RABBITMQ_SERVER="amqp://test:test@local
 
 The configuration content mentioned above needs to be configured on each node machine as needed.
 
-Then you can find the corresponding binary files data.exe and interface.exe in the release. Run the two programs on different node machines through the command line.
+Then you can find the corresponding binary files data.exe and interface.exe in the release. Run the two programs on different node machines through the command line. 
+
+**After this project version v0.2,  six data server nodes must be equipped, otherwise the service will be unavailable.**
 
 You can compile it yourself with the source code.
 

distributed/README.MD

@@ -1,13 +1,58 @@
 package config
 
 import (
+	"flag"
+	"github.com/GuoYuefei/DOStorage1/distributed/utils"
+	"gopkg.in/yaml.v2"
+	"io/ioutil"
+	"log"
 	"os"
+	"os/exec"
+	"path"
+	"path/filepath"
 )
 
+/**
+	如果是相对位置，则是相对执行文件的位置
+ */
+var ConfigFile string
+
 var Pub *SPub
 var ServerInf *SInf
 var ServerData *SData
 
+const usage string = "-c configFilePath"
+
+type ServerType = int
+const (
+	TypeSInf = iota
+	TypeSData
+)
+
+var ObjectRoot string
+var TempRoot string
+
+func Init() {
+	ObjectRoot = filepath.Join(ServerData.STORAGE_ROOT, "objects")
+	TempRoot = filepath.Join(ServerData.STORAGE_ROOT, "temp")
+	// make dir for ./data/objects
+	_, e := os.Stat(ObjectRoot)
+	if e != nil {
+		e := os.MkdirAll(ObjectRoot, os.ModePerm)
+		if e != nil {
+			log.Fatal(e)
+		}
+	}
+
+	_, e = os.Stat(TempRoot)
+	if e != nil {
+		e := os.MkdirAll(TempRoot, os.ModePerm)
+		if e != nil {
+			log.Fatal(e)
+		}
+	}
+}
+
 func init() {
 
 	Pub = &SPub{
@@ -17,44 +62,93 @@ func init() {
 	ServerInf = &SInf{
 		LISTEN_ADDRESS: "localhost:23333",
 	}
+	ServerInf.SPub = Pub
 	ServerData = &SData{
 		LISTEN_ADDRESS: "localhost:23334",
 		STORAGE_ROOT: "./data",
 	}
+	ServerData.SPub = Pub
+
+}
+
+// serverType
+// TypeSInf interface server
+// TypeSData data server
+func ConfigParse(serverType ServerType) {
+
+	switch serverType {
+	case TypeSInf:
+		interfaceConfigFile, err := ioutil.ReadFile(path.Join(ConfigFile))
+		if err == nil {
+			yaml.Unmarshal(interfaceConfigFile, ServerInf)
+		}
+		utils.Log.Printf(utils.Debug, "after read config file, %v\n", ServerInf, Pub)
+	case TypeSData:
+		dataConfigFile, err := ioutil.ReadFile(path.Join(ConfigFile))
+		if err == nil {
+			yaml.Unmarshal(dataConfigFile, ServerData)
+		}
+		utils.Log.Printf(utils.Debug, "after read config file, %v\n", ServerData, Pub)
+	default:
+		utils.Log.Println(utils.Warning, "Server Type No MATCH!")
+	}
 
 	if os.Getenv("RABBITMQ_SERVER") != "" {
 		Pub.RABBITMQ_SERVER = os.Getenv("RABBITMQ_SERVER")
 	}
 	if os.Getenv("ES_SERVER") != "" {
 		Pub.ES_SERVER = os.Getenv("ES_SERVER")
 	}
-
 	if os.Getenv("LISTEN_ADDRESS") != "" {
 		ServerData.LISTEN_ADDRESS = os.Getenv("LISTEN_ADDRESS")
 		ServerInf.LISTEN_ADDRESS = ServerData.LISTEN_ADDRESS
 	}
-
 	if os.Getenv("STORAGE_ROOT") != "" {
 		ServerData.STORAGE_ROOT = os.Getenv("STORAGE_ROOT")
 	}
+	// 最后如果STORAGE_ROOT是相对位置的话，转成绝对路径
+	if !filepath.IsAbs(ServerData.STORAGE_ROOT) {
+		exePath, _ := exec.LookPath(os.Args[0])
+		path, _ := filepath.Abs(filepath.Dir(exePath))
+		ServerData.STORAGE_ROOT = filepath.Join(path, ServerData.STORAGE_ROOT)
+	}
 
-	// todo config
-
+	// 配置结束后初始化
+	Init()
+}
 
+func Flags(serverType ServerType) {
+	switch serverType {
+	case TypeSInf:
+		flag.StringVar(&ConfigFile, "c", "./config/interface.yml", usage)
+		flag.StringVar(&ConfigFile, "-config", "./config/interface.yml", usage)
+	case TypeSData:
+		flag.StringVar(&ConfigFile, "c", "./config/data.yml", usage)
+		flag.StringVar(&ConfigFile, "-config", "./config/data.yml", usage)
+	default:
+		utils.Log.Println(utils.Warning, "Server Type No MATCH!")
+	}
+	// 相对位置转换成绝对位置
+	if !filepath.IsAbs(ConfigFile) {
+		exePath, _ := exec.LookPath(os.Args[0])
+		path, _ := filepath.Abs(filepath.Dir(exePath))
+		ConfigFile = filepath.Join(path, ConfigFile)
+	}
+	utils.Log.Println(utils.Info, "Use config file: ", ConfigFile)
 }
 
 type SPub struct {
-	ES_SERVER string
-	RABBITMQ_SERVER string
+	ES_SERVER string			`yaml:"es_server"`
+	RABBITMQ_SERVER string		`yaml:"rabbitmq_server"`
 }
 
 type SInf struct {
-	*SPub
-	LISTEN_ADDRESS string
+	*SPub						`yaml:"public"`
+	LISTEN_ADDRESS string		`yaml:"listen_address"`
 }
 
 type SData struct {
-	*SPub
-	LISTEN_ADDRESS string
-	STORAGE_ROOT string
+	*SPub						`yaml:"public"`
+	LISTEN_ADDRESS string		`yaml:"listen_address"`
+	STORAGE_ROOT string			`yaml:"storage_root"`
 }

distributed/config/config.go

@@ -0,0 +1,13 @@
+package config
+
+import (
+	"fmt"
+	"testing"
+)
+
+func Test_Init(t *testing.T) {
+	fmt.Println(Pub)
+	fmt.Printf("%p\n", Pub)
+	fmt.Println(ServerInf)
+	fmt.Println(ServerData)
+}

distributed/config/config_test.go

@@ -0,0 +1,5 @@
+public:
+  es_server: localhost:9200
+  rabbitmq_server: amqp://test:test@localhost:5672/
+listen_address: localhost:23334
+storage_root: ./data

distributed/config/data.yml

@@ -0,0 +1,4 @@
+public:
+  es_server: localhost:9200
+  rabbitmq_server: amqp://test:test@localhost:5672/
+listen_address: localhost:23333

distributed/config/interface.yml

@@ -1,51 +1,33 @@
 package locate
 
 import (
+	"fmt"
 	"github.com/GuoYuefei/DOStorage1/distributed/config"
 	"github.com/GuoYuefei/DOStorage1/distributed/rabbitmq"
+	"github.com/GuoYuefei/DOStorage1/distributed/types"
 	"github.com/GuoYuefei/DOStorage1/distributed/utils"
-	"log"
-	"os"
 	"path/filepath"
 	"strconv"
+	"strings"
 	"sync"
 )
 
 var objects map[string]int = make(map[string]int)
 var mutex sync.Mutex
 
-var ObjectRoot = filepath.Join(config.ServerData.STORAGE_ROOT, "objects")
-var TempRoot = filepath.Join(config.ServerData.STORAGE_ROOT, "temp")
-
-func init() {
-	// make dir for ./data/objects
-	_, e := os.Stat(ObjectRoot)
-	if e != nil {
-		e := os.MkdirAll(ObjectRoot, os.ModePerm)
-		if e != nil {
-			log.Fatal(e)
-		}
-	}
-
-	_, e = os.Stat(TempRoot)
-	if e != nil {
-		e := os.MkdirAll(TempRoot, os.ModePerm)
-		if e != nil {
-			log.Fatal(e)
-		}
-	}
-}
-
-func Locate(hash string) bool {
+func Locate(hash string) int {
 	mutex.Lock()
-	_, ok := objects[hash]
+	id, ok := objects[hash]
 	mutex.Unlock()
-	return ok
+	if !ok {
+		return -1
+	}
+	return id
 }
 
-func Add(hash string) {
+func Add(hash string, id int) {
 	mutex.Lock()
-	objects[hash] = 1
+	objects[hash] = id
 	mutex.Unlock()
 }
 
@@ -66,17 +48,26 @@ func StartLocate() {
 		hash, err := strconv.Unquote(string(msg.Body))
 		utils.PanicOnError(err, "Unquote error")
 		utils.Log.Printf(utils.Debug, "get hash %s for locate\n", hash)
-		if Locate(hash) {
-			// 定位成功就发送自己的信息出去
-			q.Send(msg.ReplyTo, config.ServerData.LISTEN_ADDRESS)
+		id := Locate(hash)
+		if id != -1 {
+			q.Send(msg.ReplyTo, types.LocateMessage{Addr: config.ServerData.LISTEN_ADDRESS, Id: id})
 		}
 	}
 }
 
 func CollectObjects() {
-	files, _ := filepath.Glob(filepath.Join(ObjectRoot, "/*"))
+	files, _ := filepath.Glob(filepath.Join(config.ObjectRoot, "/*"))
 	for i := range files {
-		hash := filepath.Base(files[i])
-		objects[hash] = 1
+		file := strings.Split(filepath.Base(files[i]), ".")
+		if len(file) != 3 {
+			utils.PanicOnError(fmt.Errorf("file name error, format error"), "")
+			continue
+		}
+		hash := file[0]
+		id, e := strconv.Atoi(file[1])
+		if e != nil {
+			utils.PanicOnError(fmt.Errorf("file name error. id is not a number"), "")
+		}
+		objects[hash] = id
 	}
-}
+}