package peer import ( "encoding/json" "errors" "fmt" "regexp" "strings" "cloud.o-forge.io/core/oc-lib/models/utils" "cloud.o-forge.io/core/oc-lib/tools" ) type PeerExecution struct { Method string `json:"method" bson:"method"` Url string `json:"url" bson:"url"` Body map[string]interface{} `json:"body" bson:"body"` DataType int `json:"data_type" bson:"data_type"` DataID string `json:"data_id" bson:"data_id"` } type PeerCache struct { Executions []*PeerExecution } func (p *PeerCache) urlFormat(url string, dt utils.DataType) string { if strings.Contains(url, "localhost") || strings.Contains(url, "127.0.0.1") { url = strings.ReplaceAll(url, "localhost", dt.API()) url = strings.ReplaceAll(url, "127.0.0.1", dt.API()) r := regexp.MustCompile("(:[0-9]+)") t := r.FindString(url) if t != "" { url = strings.Replace(url, t, ":8080", -1) } r.ReplaceAllString(url, ":8080") } return url } func (p *PeerCache) checkPeerStatus(peerID string, appName string, caller *tools.HTTPCaller) (*Peer, bool) { api := tools.API{} access := (&Peer{}).GetAccessor(nil) res, code, _ := access.LoadOne(peerID) if code != 200 { return nil, false } methods := caller.URLS[utils.PEER.String()] fmt.Println("PEER AFT 3", methods) if methods == nil { return res.(*Peer), false } meth := methods[tools.POST] fmt.Println("PEER AFT 4", meth) if meth == "" { return res.(*Peer), false } url := p.urlFormat(res.(*Peer).Url+meth, utils.PEER) fmt.Println("PEER AFT 5", url) state, services := api.CheckRemotePeer(url) res.(*Peer).Services = services access.UpdateOne(res, peerID) fmt.Println("PEER AFT 6", state, services, appName) return res.(*Peer), state != tools.DEAD && services[appName] == 0 } func (p *PeerCache) GetAccessor(caller *tools.HTTPCaller) utils.Accessor { data := New() data.Init(utils.PEER, caller) return data } func (p *PeerCache) LaunchPeerExecution(peerID string, dataID string, dt utils.DataType, method tools.METHOD, body map[string]interface{}, caller *tools.HTTPCaller) (*PeerExecution, error) { methods := caller.URLS[dt.String()] if _, ok := methods[method]; !ok { return nil, errors.New("no path found") } meth := methods[method] if meth == "" { return nil, errors.New("no path found") } else { meth = strings.ReplaceAll(meth, ":id", dataID) } url := "" if mypeer, ok := p.checkPeerStatus(peerID, dt.API(), caller); !ok { pexec := &PeerExecution{ Method: method.String(), Url: p.urlFormat((mypeer.Url)+meth, dt), Body: body, DataType: dt.EnumIndex(), DataID: dataID, } mypeer.AddExecution(*pexec) mypeer.GetAccessor(nil).UpdateOne(mypeer, peerID) return nil, errors.New("peer is not reachable") } else { url = p.urlFormat((mypeer.Url)+meth, dt) fmt.Println("LaunchPeerExecution AFT 3", url) tmp := mypeer.FailedExecution mypeer.FailedExecution = []PeerExecution{} mypeer.GetAccessor(nil).UpdateOne(mypeer, peerID) for _, v := range tmp { go p.exec(v.Url, tools.ToMethod(v.Method), v.Body, caller) } } return nil, p.exec(url, method, body, caller) } func (p *PeerCache) exec(url string, method tools.METHOD, body map[string]interface{}, caller *tools.HTTPCaller) error { var b []byte var err error if method == tools.POST { b, err = caller.CallPost(url, "", body) } if method == tools.GET { b, err = caller.CallGet(url, "") } if method == tools.DELETE { b, err = caller.CallDelete(url, "") } var m map[string]interface{} json.Unmarshal(b, &m) if err != nil { return err } fmt.Println("LaunchPeerExecution AFT 3", m, url) if e, ok := m["error"]; !ok && e != "" { return errors.New(fmt.Sprintf("%v", m["error"])) } return err }