| | |
| | | receiveSqlInterface = c |
| | | } |
| | | |
| | | type DbHandler interface { |
| | | Execute(sql string) bool |
| | | } |
| | | |
| | | type DbDumpHandler interface { |
| | | Dump(tables string) string |
| | | } |
| | | |
| | | //数据库execute句柄 |
| | | 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" |
| | |
| | | //fmt.Println(string(ev.Payload)) |
| | | var tableNames []string |
| | | err := json.Unmarshal(ev.Payload, &tableNames) |
| | | if err !=nil { |
| | | //fmt.Println("Query tableNames unmarshal err") |
| | | return |
| | | } |
| | | fmt.Println("androidSync tableNames:", tableNames, "err:", err) |
| | | |
| | | rows, err := ExecuteQuerySql(tableNames) |
| | | if err != nil { |
| | | //fmt.Println("err: ", err) |
| | | return |
| | | } |
| | | var rowsReturn []Rows |
| | | for _, r := range rows { |
| | | rowsReturn = append(rowsReturn, *r) |
| | | } |
| | | dumpData := dbDumpH.Dump(strings.Join(tableNames, ",")) |
| | | |
| | | bytesReturn, err := json.Marshal(rowsReturn) |
| | | //fmt.Println("results: ", bytesReturn) |
| | | bytesReturn, _ := json.Marshal(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) |
| | |
| | | func (a *Agent) GetTableDataFromCluster(tableNames []string) (*[]string, error) { |
| | | //members: get name of first member |
| | | mbs := a.GroupMembers(a.conf.ClusterID) |
| | | var specmembername string |
| | | specmembername := "" |
| | | for _, m := range mbs { |
| | | //fmt.Println("m", m) |
| | | if m.Name != a.conf.NodeName { //前缀:DSVAD:分析服务器 DSPAD:进出入pad |
| | | if strings.HasPrefix(a.conf.NodeName, "DSVAD"){ |
| | | if strings.HasPrefix(m.Name, "DSVAD") { |
| | | if strings.HasPrefix(a.conf.NodeName, "PSPAD"){ |
| | | if strings.HasPrefix(m.Name, "PSPAD") { |
| | | specmembername = m.Name |
| | | break |
| | | } |
| | | }else{ |
| | | specmembername = m.Name |
| | | break |
| | | } |
| | | } |
| | | } |
| | | //if specmembername == "" { |
| | | // return nil, errors.New("other pad node not found in memberlist") |
| | | //} |
| | | //fmt.Println("mbs:", mbs, "a.conf.BindAddr:", a.conf.BindAddr, "specmembername:", specmembername) |
| | | |
| | | //query: get db file. |
| | |
| | | |
| | | resp, err := a.Query(QueryEventUpdateDBData, tBytes, ¶ms) |
| | | 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) |
| | | fmt.Println("Query.resp.err:", err, "resp:", resp) |
| | | |
| | | var wg sync.WaitGroup |
| | | wg.Add(1) |