add sync DbTablePerson UserEvent

liuxiaolong

2019-08-19 c1ed6226183527f60bf7a9f15b780a2c8f67e69a

 agent.go

@@ -40,6 +40,8 @@
const (
   QueryEventGetDB        = "GetDatabase"
   QueryEventUpdateDBData = "UpdateDBData"
   UserEventSyncSql      = "SyncSql"
   UserEventSyncDbTablePersonCache      = "SyncCache"
)

// Agent warps the serf agent
@@ -110,6 +112,8 @@
   go a.BroadcastMemberlist(BroadcastInterval * time.Second)
}

var SyncDbTablePersonCacheChan = make(chan []byte,0)

// 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",
@@ -118,7 +122,7 @@

   switch ev := event.(type) {
   case serf.UserEvent:
      //fmt.Println(string(ev.Payload))
      if ev.Name == UserEventSyncSql {
      var sqlUe SqlUserEvent
      err := json.Unmarshal(ev.Payload, &sqlUe)
      if err !=nil {
@@ -130,6 +134,10 @@
         flag, _ := ExecuteSqlByGorm(sqlUe.Sql)
         fmt.Println("userEvent exec ",sqlUe.Sql,",Result:",flag)
      }
      } else if ev.Name == UserEventSyncDbTablePersonCache {
         SyncDbTablePersonCacheChan <- ev.Payload
      }


   case *serf.Query:

@@ -501,12 +509,20 @@
      fmt.Println("sqlUE marshal err:",err)
      return
   }
   err = a.UserEvent("SyncSql", ueB, false)
   err = a.UserEvent(UserEventSyncSql, ueB, false)
   if err == nil || !strings.Contains(err.Error(), "cannot contain") {
      fmt.Println("err: ", err)
   }
}

//更新同步库的比对缓存
func (a *Agent) SyncDbTablePersonCache(b []byte) {
   err := a.UserEvent(UserEventSyncDbTablePersonCache, b, false)
   if err !=nil{
      fmt.Println("UserEventSyncDbTablePersonCache err:",err)
   }
}

//Init serf Init
func Init(clusterID string, password string, nodeID string, ips []string) (*Agent, error) {
   agent, err := InitNode(clusterID, password, nodeID)