deal EventMemberLeave

liuxiaolong

2019-08-27 940e9644d5b72c5083eacf28c65eb07154d87f67

 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,15 +122,22 @@

   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)
      if ev.Name == UserEventSyncSql {
         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)
         }
      } else if ev.Name == UserEventSyncDbTablePersonCache {
         SyncDbTablePersonCacheChan <- ev.Payload
      }


   case *serf.Query:

@@ -197,6 +208,18 @@
         //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+"'"
            ExecuteSqlByGorm([]string{ leaveSql })

            fmt.Println("EventMemberLeave,current Members:",ev.Members)
         }
         return
      }


   default:
      fmt.Printf("Unknown event type: %s\n", ev.EventType().String())
@@ -481,12 +504,34 @@

}

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(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)
   }
}

@@ -556,8 +601,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!")
   }