From df17d827d53a89a19c43816d3a633ba49f9105ae Mon Sep 17 00:00:00 2001
From: liuxiaolong <736321739@qq.com>
Date: 星期三, 14 八月 2019 14:32:50 +0800
Subject: [PATCH] UserEvent exec sql use gorm

---
 agent.go |  115 +++++++++++++++++++++++++++++++++++++++++++++++----------
 1 files changed, 94 insertions(+), 21 deletions(-)

diff --git a/agent.go b/agent.go
index 78af264..8764201 100644
--- a/agent.go
+++ b/agent.go
@@ -25,6 +25,7 @@
 	"io/ioutil"
 	"net"
 	"os"
+	"sync"
 
 	//"os"
 	"strings"
@@ -118,14 +119,11 @@
 	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)
-		}
+		var sqlArr []string
+		sqlArr = append(sqlArr, string(ev.Payload))
+		//results, err := ExecuteWriteSql(sqlArr)
+		flag, _ := ExecuteSqlByGorm(sqlArr)
+		fmt.Println("userEvent exec ",sqlArr,",Result:",flag)
 
 	case *serf.Query:
 
@@ -159,22 +157,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,6 +414,70 @@
 	}()
 }
 
+//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
+	//	}
+	//}
+
+}
+
 //SyncSql boardcast sql to cluster
 func (a *Agent) SyncSql(sqlOp string) {
 	// event : use to send command to operate db.

--
Gitblit v1.8.0