From f1cd5a5e3adc85f80ba79b4b75348651865de8c7 Mon Sep 17 00:00:00 2001
From: liuxiaolong <736321739@qq.com>
Date: 星期四, 15 八月 2019 17:21:34 +0800
Subject: [PATCH] rm a.Stop when Join failed

---
 agent.go |  143 +++++++++++++++++++++++++++++++++++++++--------
 1 files changed, 118 insertions(+), 25 deletions(-)

diff --git a/agent.go b/agent.go
index 78af264..bc2585f 100644
--- a/agent.go
+++ b/agent.go
@@ -25,6 +25,7 @@
 	"io/ioutil"
 	"net"
 	"os"
+	"sync"
 
 	//"os"
 	"strings"
@@ -117,14 +118,17 @@
 
 	switch ev := event.(type) {
 	case serf.UserEvent:
-		fmt.Println(string(ev.Payload))
-		var tmpstringslice []string
-		tmpstringslice = append(tmpstringslice, string(ev.Payload))
-		fmt.Println(tmpstringslice)
-		results, err := ExecuteWriteSql(tmpstringslice)
-
-		for _, result := range results {
-			fmt.Println(result, "results err: ", err)
+		//fmt.Println(string(ev.Payload))
+		var sqlUe SqlUserEvent
+		err := json.Unmarshal(ev.Payload, &sqlUe)
+		if err !=nil {
+			fmt.Println("sqlUe unmarshal err:",err)
+			return
+		}
+		if sqlUe.Owner != a.conf.NodeName {
+			//results, err := ExecuteWriteSql(sqlArr)
+			flag, _ := ExecuteSqlByGorm(sqlUe.Sql)
+			fmt.Println("userEvent exec ",sqlUe.Sql,",Result:",flag)
 		}
 
 	case *serf.Query:
@@ -159,22 +163,33 @@
 				}
 			}
 		} else if ev.Name == QueryEventUpdateDBData {
-			fmt.Println(string(ev.Payload))
-			var tmpstringslice []string
-			tmpstringslice = append(tmpstringslice, string(ev.Payload))
-			fmt.Println(tmpstringslice)
-			rows, err := ExecuteQuerySql(tmpstringslice)
-			if err != nil {
-				fmt.Println("err: ", err)
+			//fmt.Println(string(ev.Payload))
+			//var tmpstringslice []string
+			//tmpstringslice = append(tmpstringslice, string(ev.Payload))
+			//fmt.Println(tmpstringslice)
+			//rows, err := ExecuteQuerySql(tmpstringslice)
+			//if err != nil {
+			//	fmt.Println("err: ", err)
+			//	return
+			//}
+			//var rowsReturn []Rows
+			//for _, r := range rows {
+			//	rowsReturn = append(rowsReturn, *r)
+			//}
+			var tableNames []string
+			err := json.Unmarshal(ev.Payload, &tableNames)
+			if err !=nil {
+				fmt.Println("Query tableNames unmarshal err")
 				return
 			}
-			var rowsReturn []Rows
-			for _, r := range rows {
-				rowsReturn = append(rowsReturn, *r)
+			fmt.Println("Query tableNames:",tableNames)
+			datas, err := ExecuteQueryByGorm(tableNames)
+			if err !=nil {
+				fmt.Println("queryByGorm err")
+				return
 			}
-
-			bytesReturn, err := json.Marshal(rowsReturn)
-			fmt.Println("results: ", bytesReturn)
+			bytesReturn, err := json.Marshal(datas)
+			fmt.Println("results.len: ", len(bytesReturn))
 			if query, ok := event.(*serf.Query); ok {
 				if err := query.Respond(bytesReturn); err != nil {
 					fmt.Printf("err: %s\n", err)
@@ -405,10 +420,88 @@
 	}()
 }
 
+//GetDbFromCluster get the newest database after join cluster
+//dbPathWrite the path where to write after got a database,
+func (a *Agent) GetTableDataFromCluster(tableNames []string) (*[]string,error) {
+	//members: get name of first member
+	mbs := a.GroupMembers(a.conf.ClusterID)
+	var specmembername string
+	for _, m := range mbs {
+		fmt.Println("m",m)
+		if m.Name != a.conf.NodeName {
+			specmembername = m.Name
+			break
+		}
+	}
+	fmt.Println("mbs:",mbs,"a.conf.BindAddr:",a.conf.BindAddr,"specmembername:",specmembername)
+
+	//query: get db file.
+	params := serf.QueryParam{
+		FilterNodes: strings.Fields(specmembername),
+	}
+
+	//SQL
+	tBytes, _ := json.Marshal(tableNames)
+
+	resp, err := a.Query(QueryEventUpdateDBData, tBytes, &params)
+	if err == nil || !strings.Contains(err.Error(), "cannot contain") {
+		fmt.Println("err: ", err)
+	}
+	fmt.Println("Query.resp.err:",err,"resp:",resp)
+
+	var wg sync.WaitGroup
+	wg.Add(1)
+	var dumpSqls []string
+	go func() {
+		defer wg.Done()
+		respCh := resp.ResponseCh()
+		for {
+			select {
+			case r := <-respCh:
+				fmt.Println("Query response's len:", len(r.Payload))
+				err := json.Unmarshal(r.Payload, &dumpSqls)
+				if err ==nil {
+					fmt.Println("dumpSql:",dumpSqls)
+					fmt.Println("data dump success")
+				}
+				return
+			}
+		}
+	}()
+	wg.Wait()
+	return &dumpSqls,nil
+
+	//r, err = c.Query([]string{query}, false, false)
+	//if err != nil {
+	//	return err
+	//}
+	//for _, x := range r[0].Values {
+	//	y := fmt.Sprintf("%s;\n", x[0].(string))
+	//	if _, err := w.Write([]byte(y)); err != nil {
+	//		return err
+	//	}
+	//}
+
+}
+
+type SqlUserEvent struct {
+	Owner string `json:"owner"`
+	Sql []string `json:"sql"`
+}
+
 //SyncSql boardcast sql to cluster
-func (a *Agent) SyncSql(sqlOp string) {
+func (a *Agent) SyncSql(sqlOp []string) {
 	// event : use to send command to operate db.
-	err := a.UserEvent("SyncSql", []byte(sqlOp), false)
+	var sqlUe = SqlUserEvent{
+		Owner: a.conf.NodeName,
+		Sql: sqlOp,
+	}
+	ueB, err := json.Marshal(sqlUe)
+	if err !=nil {
+		fmt.Println("sqlUE marshal err:",err)
+		return
+	}
+	err = a.UserEvent("SyncSql", ueB, false)
 	if err == nil || !strings.Contains(err.Error(), "cannot contain") {
 		fmt.Println("err: ", err)
 	}
@@ -480,8 +573,8 @@
 
 	n, err := a.Agent.Join(nodes, true)
 	if err != nil || n == 0 {
-		a.Stop()
-		fmt.Println("Stop node")
+		//a.Stop()
+		//fmt.Println("Stop node")
 		return fmt.Errorf("Error Encrypt Key!")
 	}
 

--
Gitblit v1.8.0