From a79ff0adc62e2d9e3e198d7299e7670063201607 Mon Sep 17 00:00:00 2001
From: liuxiaolong <736321739@qq.com>
Date: 星期四, 14 五月 2020 20:48:39 +0800
Subject: [PATCH] rm no use import

---
 agent.go |  227 ++++++++++++++++++++++++++++++++++----------------------
 1 files changed, 139 insertions(+), 88 deletions(-)

diff --git a/agent.go b/agent.go
index f6657d6..f5bb23b 100644
--- a/agent.go
+++ b/agent.go
@@ -61,15 +61,15 @@
 }
 
 // Create create serf agent with config
-func Create(conf *Config) (*Agent, error) {
+func Create(conf *Config, snapshotPath string) (*Agent, error) {
 	// config cover to serf config
-	serfConf, err := conf.convertToSerf()
+	serfConf, err := conf.convertToSerf(snapshotPath)
 	if err != nil {
 		return nil, err
 	}
 
 	// create serf agent with serf config
-	fmt.Println("conf.Config.EncryptKey:", conf.EncryptKey)
+	//fmt.Println("conf.Config.EncryptKey:", conf.EncryptKey)
 	serfAgent, err := agent.Create(conf.Config, serfConf, nil)
 	if err != nil {
 		return nil, err
@@ -77,12 +77,12 @@
 	// Create the keyring
 	keyring, err := memberlist.NewKeyring(nil, []byte(conf.EncryptKey))
 	if err != nil {
-		fmt.Printf("Failed to restore keyring: %s", err)
+		//fmt.Printf("Failed to restore keyring: %s", err)
 		return nil, err
 	}
 	serfConf.MemberlistConfig.Keyring = keyring
-	fmt.Printf("[INFO] agent: Restored keyring with %d keys from %s",
-		len(conf.EncryptKey), conf.EncryptKey)
+	//fmt.Printf("[INFO] agent: Restored keyring with %d keys from %s",
+	//	len(conf.EncryptKey), conf.EncryptKey)
 
 	return &Agent{
 		Agent:   serfAgent,
@@ -94,6 +94,9 @@
 
 // Start agent
 func (a *Agent) Start(ctx context.Context) {
+	log.Println("aaaaaaaaa")
+	//log.Println(string(debug.Stack()))
+	//debug.PrintStack()
 	err := a.Agent.Start()
 	if err != nil {
 		log.Println(err, "start serf agent failed")
@@ -113,6 +116,40 @@
 	go a.BroadcastMemberlist(BroadcastInterval * time.Second)
 }
 
+// define call back interface
+type ReceiveSqlInterface interface {
+	Forward(content string)
+}
+
+// callback instance
+var receiveSqlInterface ReceiveSqlInterface
+
+// save the callback instance
+func RegisterReceiveSqlInterface(c ReceiveSqlInterface) {
+	receiveSqlInterface = c
+}
+
+type DbHandler interface {
+	Execute(sql string) bool
+}
+
+type DbDumpHandler interface {
+	Dump(tables string) string
+}
+
+//鏁版嵁搴揺xecute鍙ユ焺
+var DbHandle DbHandler
+//鏁版嵁搴撳浠藉彞鏌�,鍙湪syncdb涓敤鍒�
+var dbDumpH DbDumpHandler
+
+func RegisterDbHandler(h DbHandler) {
+	DbHandle = h
+}
+
+func RegisterDbDumpHandler(h DbDumpHandler) {
+	dbDumpH = h
+}
+
 // HandleEvent Handles serf.EventMemberJoin events,
 // which will wait for members to join until the number of group members is equal to "groupExpect"
 // when the startup mode is "ModeCluster",
@@ -123,16 +160,16 @@
 	case serf.UserEvent:
 		if ev.Name == UserEventSyncSql {
 			var sqlUe SqlUserEvent
-			fmt.Println(string(ev.Payload))
+			//fmt.Println(string(ev.Payload))
 			err := json.Unmarshal(ev.Payload, &sqlUe)
 			if err != nil {
-				fmt.Println("sqlUe unmarshal err:", err)
+				//fmt.Println("sqlUe unmarshal err:", err)
 				return
 			}
 			if sqlUe.Owner != a.conf.NodeName {
-				results, _ := ExecuteWriteSql(sqlUe.Sql, true)
-				//flag, _ := ExecuteSqlByGorm(sqlUe.Sql)
-				fmt.Println("userEvent exec ", sqlUe.Sql, ",Result:", results)
+				if receiveSqlInterface != nil && sqlUe.Sql !=nil && len(sqlUe.Sql) > 0 {
+					receiveSqlInterface.Forward(sqlUe.Sql[0])
+				}
 			}
 		}
 
@@ -142,76 +179,67 @@
 			//bak file and send resp
 			filename, err := BakDbFile()
 			if err != nil {
-				fmt.Println("bak db file error!")
+				//fmt.Println("bak db file error!")
 				return
 			}
-			fmt.Println(filename)
+			//fmt.Println(filename)
 
 			filebuf, err := ioutil.ReadFile(filename)
-			fmt.Println("filebuf: ", len(filebuf))
+			//fmt.Println("filebuf: ", len(filebuf))
 			if err != nil {
-				fmt.Printf("file to []bytes error: %s\n", err)
+				//fmt.Printf("file to []bytes error: %s\n", err)
 				return
 			}
 
 			err = os.Remove(filename)
 			if err != nil {
-				fmt.Printf("remove file%s\n failed", filename)
+				//fmt.Printf("remove file%s\n failed", filename)
 				return
 			}
 
-			fmt.Println("query payload: ", len(ev.Payload))
+			//fmt.Println("query payload: ", len(ev.Payload))
 			if query, ok := event.(*serf.Query); ok {
 				if err := query.Respond(filebuf); err != nil {
-					fmt.Printf("err: %s\n", err)
+					//fmt.Printf("err: %s\n", err)
 					return
 				}
 			}
 		} 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)
-				return
-			}
-			var rowsReturn []Rows
-			for _, r := range rows {
-				rowsReturn = append(rowsReturn, *r)
-			}
+			//fmt.Println(string(ev.Payload))
+			var tableNames []string
+			err := json.Unmarshal(ev.Payload, &tableNames)
+			fmt.Println("androidSync tableNames:", tableNames, "err:", err)
 
-			bytesReturn, err := json.Marshal(rowsReturn)
-			fmt.Println("results: ", bytesReturn)
+			dumpData := dbDumpH.Dump(strings.Join(tableNames, ","))
+
+			bytesReturn := []byte(dumpData)
+			fmt.Println("androidSync len(bytesReturn): ", len(bytesReturn))
 			if query, ok := event.(*serf.Query); ok {
 				if err := query.Respond(bytesReturn); err != nil {
-					fmt.Printf("err: %s\n", err)
+					fmt.Println("androidSync query.Respond err:", err)
 					return
+				} else {
+					fmt.Println("androidSync query.Respond success")
 				}
 			}
-
-			//var res []*Rows
-			//json.Unmarshal(bytesReturn, &res)
 		}
 	case serf.MemberEvent:
 		if event.EventType() == serf.EventMemberLeave {
 			if ev.Members != nil && len(ev.Members) == 1 {
 				leaveMember := ev.Members[0]
 				leaveSql := "delete from cluster_node where node_id='" + leaveMember.Name + "'"
-				fmt.Println("leaveSql:", leaveSql)
-				_, err := ExecuteWriteSql([]string{leaveSql}, false)
-				if err != nil {
-					fmt.Printf("err: %s\n", err)
+				//fmt.Println("leaveSql:", leaveSql)
+				if !DbHandle.Execute(leaveSql) {
+					fmt.Println("DbHandle.Execute ret false")
 				}
 
-				fmt.Println("EventMemberLeave,current Members:", ev.Members)
+				//fmt.Println("EventMemberLeave,current Members:", ev.Members)
 			}
 			return
 		}
 
 	default:
-		fmt.Printf("Unknown event type: %s\n", ev.EventType().String())
+		//fmt.Printf("Unknown event type: %s\n", ev.EventType().String())
 	}
 }
 
@@ -220,7 +248,7 @@
 	serf := a.Agent.Serf()
 	mb := serf.LocalMember()
 	mblist := serf.Memberlist()
-	fmt.Println("mb:", mb)
+	//fmt.Println("mb:", mb)
 
 	// copy local node
 	localNode := *mblist.LocalNode()
@@ -241,7 +269,7 @@
 	//localNode.Addr = net.IPv4(255,255,255,255)
 	localNode.Port = BroadcastPort
 	for {
-		// fmt.Printf("localNode: %v %v\n", nodeName, nodeAddress)
+		// //fmt.Printf("localNode: %v %v\n", nodeName, nodeAddress)
 		mblist.SendBestEffort(&localNode, []byte(message))
 		time.Sleep(delay)
 	}
@@ -379,7 +407,7 @@
 			break
 		}
 	}
-	fmt.Println(specmembername)
+	//fmt.Println(specmembername)
 
 	//query: get db file.
 	params := serf.QueryParam{
@@ -388,7 +416,7 @@
 
 	resp, err := a.Query(QueryEventGetDB, []byte(""), &params)
 	if err == nil || !strings.Contains(err.Error(), "cannot contain") {
-		fmt.Println("err: ", err)
+		//fmt.Println("err: ", err)
 	}
 
 	go func() {
@@ -396,18 +424,18 @@
 		for {
 			select {
 			case r := <-respCh:
-				fmt.Println("x length is: ", len(r.Payload))
+				//fmt.Println("x length is: ", len(r.Payload))
 
 				// // byte to file.
 				SerfDbConn.Close()
 				SerfDbConn = nil
 				err = ioutil.WriteFile(dbPathWrite, r.Payload, 0644)
 				if err != nil {
-					fmt.Println("query byte to file error!", err)
+					//fmt.Println("query byte to file error!", err)
 				}
 				err := InitDbConn("")
 				if err != nil {
-					fmt.Println("create db conn of test.db error: ", err)
+					//fmt.Println("create db conn of test.db error: ", err)
 				}
 				return
 			}
@@ -417,29 +445,18 @@
 
 //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) {
+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 { //鍓嶇紑锛欴SVAD:鍒嗘瀽鏈嶅姟鍣� DSPAD:杩涘嚭鍏ad
-			if strings.HasPrefix(a.conf.NodeName, "DSVAD"){
-				if strings.HasPrefix(m.Name, "DSVAD") {
-					specmembername = m.Name
-					break
-				}
-			}else{
-				specmembername = m.Name
-				break
-			}
-		}
+	targetName := a.GetOneAliveNodeId(20 * time.Second)
+
+	if targetName == "" {
+		return nil, errors.New("alive node not found in cluster")
 	}
-	fmt.Println("mbs:", mbs, "a.conf.BindAddr:", a.conf.BindAddr, "specmembername:", specmembername)
+	//fmt.Println("mbs:", mbs, "a.conf.BindAddr:", a.conf.BindAddr, "targetName:", targetName)
 
 	//query: get db file.
 	params := serf.QueryParam{
-		FilterNodes: strings.Fields(specmembername),
+		FilterNodes: strings.Fields(targetName),
 	}
 
 	//SQL
@@ -447,13 +464,13 @@
 
 	resp, err := a.Query(QueryEventUpdateDBData, tBytes, &params)
 	if err == nil || !strings.Contains(err.Error(), "cannot contain") {
-		fmt.Println("err: ", err)
+		//fmt.Println("QueryEventUpdateDBData err: ")
 	}
 	fmt.Println("Query.resp.err:", err, "resp:", resp)
 
 	var wg sync.WaitGroup
 	wg.Add(1)
-	var dumpSqls []string
+	var dumpSqls string
 	go func() {
 		defer wg.Done()
 		respCh := resp.ResponseCh()
@@ -461,10 +478,11 @@
 			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)
+				dumpSqls = string(r.Payload)
+				if len(dumpSqls) >0 {
 					fmt.Println("data dump success")
+				} else {
+					fmt.Println("Query response.len = 0")
 				}
 				return
 			}
@@ -472,6 +490,39 @@
 	}()
 	wg.Wait()
 	return &dumpSqls, nil
+}
+
+func (a *Agent) GetOneAliveNodeId(timeOut time.Duration) string {
+	nodeId := ""
+	sTk := time.NewTicker(1 * time.Second)
+	outTk := time.NewTicker(timeOut)
+	defer sTk.Stop()
+	defer outTk.Stop()
+
+	for {
+		select {
+		case <- sTk.C:
+			mbs := a.GroupMembers(a.conf.ClusterID)
+			for _, m := range mbs {
+				//fmt.Println("m", m)
+				if m.Name != a.conf.NodeName { //鍓嶇紑锛欴SVAD:鍒嗘瀽鏈嶅姟鍣� DSPAD:杩涘嚭鍏ad
+					if strings.HasPrefix(a.conf.NodeName, "PAD"){
+						if strings.HasPrefix(m.Name, "PAD") {
+							nodeId = m.Name
+							goto Out
+						}
+					}
+				}
+			}
+		case <-outTk.C:
+			fmt.Println("cannot get alive node from cluster,time out")
+			goto Out
+		default:
+
+		}
+	}
+Out:
+	return nodeId
 }
 
 type SqlUserEvent struct {
@@ -488,26 +539,26 @@
 	}
 	ueB, err := json.Marshal(sqlUe)
 	if err != nil {
-		fmt.Println("sqlUE marshal err:", err)
+		//fmt.Println("sqlUE marshal err:", err)
 		return
 	}
 	err = a.UserEvent(UserEventSyncSql, ueB, false)
 	if err == nil || !strings.Contains(err.Error(), "cannot contain") {
-		fmt.Println("err: ", err)
+		//fmt.Println("err: ", err)
 	}
 }
 
 //Init serf Init
-func Init(clusterID string, password string, nodeID string, addrs []string) (*Agent, error) {
-	agent, err := InitNode(clusterID, password, nodeID)
+func Init(clusterID string, password string, nodeID string, addrs []string, snapshotPath string) (*Agent, error) {
+	agent, err := InitNode(clusterID, password, nodeID, snapshotPath)
 	if err != nil {
-		fmt.Printf("InitNode failed, error: %s", err)
+		//fmt.Printf("InitNode failed, error: %s", err)
 		return agent, err
 	}
 
 	err = agent.JoinByNodeAddrs(addrs)
 	if err != nil {
-		fmt.Printf("JoinByNodeIP failed, error: %s", err)
+		//fmt.Printf("JoinByNodeIP failed, error: %s", err)
 		return agent, err
 	}
 
@@ -515,9 +566,9 @@
 }
 
 //InitNode web鍚庡彴鏀跺埌鍒涘缓闆嗙兢鐨勮姹傦紝
-func InitNode(clusterID string, password string, nodeID string) (*Agent, error) {
+func InitNode(clusterID string, password string, nodeID string, snapshotPath string) (*Agent, error) {
 	conf := DefaultConfig()
-	fmt.Println("clusterID:", clusterID, "password:", password, "nodeID:", nodeID)
+	//fmt.Println("clusterID:", clusterID, "password:", password, "nodeID:", nodeID)
 	conf.ClusterID = clusterID
 	conf.NodeName = nodeID
 	if password == "" {
@@ -531,9 +582,9 @@
 		}
 		conf.EncryptKey = password
 	}
-	agent, err := Create(conf)
+	agent, err := Create(conf, snapshotPath)
 	if err != nil {
-		fmt.Printf("create agent failed, error: %s", err)
+		//fmt.Printf("create agent failed, error: %s", err)
 		return agent, err
 	}
 
@@ -543,9 +594,9 @@
 		agent.ShutdownCh()
 	}()
 	time.Sleep(time.Second)
-	fmt.Println("Stats:", agent.Agent.Serf().Stats())
-	fmt.Println("EncryptionEnabled:", agent.Agent.Serf().EncryptionEnabled())
-	fmt.Println("create agent sucess!!")
+	//fmt.Println("Stats:", agent.Agent.Serf().Stats())
+	//fmt.Println("EncryptionEnabled:", agent.Agent.Serf().EncryptionEnabled())
+	//fmt.Println("create agent sucess!!")
 
 	return agent, nil
 }
@@ -586,7 +637,7 @@
 
 func (a *Agent) GetNodes() (nodes []NodeInfo) {
 	var node NodeInfo
-	fmt.Println("a.conf.ClusterID:", a.conf.ClusterID)
+	//fmt.Println("a.conf.ClusterID:", a.conf.ClusterID)
 	mbs := a.GroupMembers(a.conf.ClusterID)
 	for _, mb := range mbs {
 		node.NodeID = mb.Name
@@ -598,4 +649,4 @@
 	}
 
 	return nodes
-}
+}
\ No newline at end of file

--
Gitblit v1.8.0