From 2ea23e2364f742f528fb8b97657212c174d1bbd7 Mon Sep 17 00:00:00 2001
From: zhangqian <zhangqian@123.com>
Date: 星期六, 21 十月 2023 14:06:59 +0800
Subject: [PATCH] 增加切换serf状态接口（调试用）

---
 serf/sync.go |   91 ++++++++++++++++++++++++++++++++++++++++++---
 1 files changed, 84 insertions(+), 7 deletions(-)

diff --git a/serf/sync.go b/serf/sync.go
index 7675e84..829d412 100644
--- a/serf/sync.go
+++ b/serf/sync.go
@@ -1,6 +1,7 @@
 package serf
 
 import (
+	"apsClient/pkg/logx"
 	"context"
 	"encoding/json"
 	"fmt"
@@ -19,7 +20,10 @@
 )
 
 var (
-	agent = SyncServer{}
+	agent       = SyncServer{}
+	dependProcs = []string{
+		bhomeclient.Proc_System_Service,
+	}
 )
 
 const (
@@ -101,6 +105,42 @@
 	bhomedbapi.InitDoReq(client.RequestOnly)
 	//bhomedbapi.InitLog(logger.Debug)
 
+	// 闇�瑕佺瓑寰卻ystem-service杩涚▼鎴愬姛鍚姩鍚庯紝鎵嶈兘鑾峰彇闆嗙兢鐘舵��(鎴栬�呬繚璇佺▼搴忓惎鍔ㄦ椂鑾峰彇鍒版纭殑鐘舵��)
+	tryTimes := 0
+loop:
+	for {
+		select {
+		case <-q:
+			initChan <- false
+			return
+		default:
+			if tryTimes < 15 {
+				clients, err := client.GetRegisteredClient()
+				if err == nil && len(clients) > 0 {
+					var existingProcs []string
+					for _, c := range clients {
+						if c.Online {
+							existingProcs = append(existingProcs, string(c.Proc.ProcId))
+						}
+					}
+					if diff := arrayContains(existingProcs, dependProcs); diff == "" {
+						break loop
+					} else {
+						logx.Errorf("Proc: %s is not running!", diff)
+						time.Sleep(time.Second * 1)
+					}
+				} else {
+					tryTimes++
+					time.Sleep(time.Second * 5)
+				}
+			} else {
+				logx.Errorf("tried 15 times, client.GetRegisteredClient failed")
+				initChan <- false
+				return
+			}
+		}
+	}
+
 	go client.StartServer(nil)
 
 	ss.bhClient = client
@@ -157,7 +197,7 @@
 		return err
 	}
 
-	fmt.Println("鍔犲叆闆嗙兢, 璇锋眰鍚屾鍏ㄩ噺鏁版嵁,id:", ss.ServerId)
+	logx.Debugf("鍔犲叆闆嗙兢, 璇锋眰鍚屾鍏ㄩ噺鏁版嵁,id:%s", ss.ServerId)
 	return ss.bhClient.Publish(serfSyncTopic, b)
 }
 
@@ -177,7 +217,7 @@
 			// 澶勭悊鍚屾鍏ㄩ噺鏁版嵁鐨勮姹�
 			if string(busMsg.Topic) == ss.queryTableTopic {
 				if ss.ClusterStatus == "master" {
-					fmt.Println("鎺ユ敹鍒板悓姝ュ叏閲忔暟鎹姹�")
+					logx.Debugf("鎺ユ敹鍒板悓姝ュ叏閲忔暟鎹姹�.")
 					ss.handleSyncTableMessage(busMsg.Data)
 				}
 			}
@@ -248,7 +288,7 @@
 
 		err = tx.Exec(delSql).Error
 		if err != nil {
-			fmt.Println("鍒犻櫎鏈湴鐨勫悓姝ュ簱鏁版嵁澶辫触,", err.Error())
+			logx.Errorf("鍒犻櫎鏈湴鐨勫悓姝ュ簱鏁版嵁澶辫触, %s", err.Error())
 		}
 	}
 
@@ -285,7 +325,7 @@
 
 	ss.ClusterStatus = reply.Msg
 
-	fmt.Println("褰撳墠闆嗙兢鐘舵��:", ss.ClusterStatus)
+	logx.Debugf("褰撳墠闆嗙兢鐘舵��: %s", ss.ClusterStatus)
 
 	return reply.Msg
 }
@@ -362,9 +402,12 @@
 	}
 }
 
+// serf 鍚屾鏁版嵁鐨勯檺鍒朵负92160 byte
 func (ss *SyncServer) handleSyncTableMessage(msg []byte) error {
+	sizeLimit := 61440
 	targetId := string(msg)
-	fmt.Println("鍚屾鍏ㄩ噺鏁版嵁缁欒妭鐐�:", targetId)
+
+	//fmt.Println("鍚屾鍏ㄩ噺鏁版嵁缁欒妭鐐�:", targetId)
 	sqls, err := DumpTables(ss.sqlDB, ss.syncTables)
 	if err != nil {
 		fmt.Println("DumpTables error, ", err.Error())
@@ -372,7 +415,41 @@
 	}
 
 	syncSql := strings.Join(sqls, ";")
-	err = ss.pubSyncSqlMessage([]byte(syncSql), targetId)
+	if len(syncSql) < sizeLimit {
+		err = ss.pubSyncSqlMessage([]byte(syncSql), targetId)
+	} else {
+		shard := ""
+		for _, sql := range sqls{
+			if len(shard) + len(sql) > sizeLimit {
+				err = ss.pubSyncSqlMessage([]byte(shard), targetId)
+				shard = ""
+			}
+
+			shard = fmt.Sprintf("%s%s;",shard,sql)
+		}
+	}
 
 	return err
 }
+
+func arrayContains(list []string, arr []string) string {
+	if arr == nil || list == nil {
+		return ""
+	}
+
+	for _, s := range arr {
+		isExist := false
+		for _, t := range list {
+			if s == t {
+				isExist = true
+				break
+			}
+		}
+
+		if !isExist {
+			return s
+		}
+	}
+
+	return ""
+}

--
Gitblit v1.8.0