oc-lib/models/peer/peer_cache.go
2024-08-23 08:23:24 +02:00

154 lines
3.6 KiB
Go

package peer
import (
"encoding/json"
"errors"
"fmt"
"regexp"
"strings"
"time"
"cloud.o-forge.io/core/oc-lib/logs"
"cloud.o-forge.io/core/oc-lib/models/utils"
"cloud.o-forge.io/core/oc-lib/tools"
)
var currentRountine = 0
var singleton = &PeerCache{
Executions: []*PeerExecution{},
}
type PeerExecution struct {
Method tools.METHOD
Url string
Body map[string]interface{}
IsMySelf bool
Caller tools.HTTPCaller
PeerID string
DataType utils.DataType
DataID string
}
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, caller *tools.HTTPCaller) bool {
api := tools.API{}
access := (&Peer{}).GetAccessor(nil)
res, code, _ := access.LoadOne(peerID)
if code != 200 {
return false
}
methods := caller.URLS[utils.PEER.String()]
if methods == nil {
return false
}
meth := methods[tools.POST]
if meth == "" {
return false
}
url := p.urlFormat(res.(*Peer).Url+meth, utils.PEER)
state := api.CheckRemotePeer(url)
res.(*Peer).State = state
access.UpdateOne(res, peerID)
fmt.Printf("Peer %v is %v\n", peerID, state)
return state != tools.DEAD
}
func (p *PeerCache) GetAccessor(caller *tools.HTTPCaller) utils.Accessor {
data := New()
data.Init(utils.PEER, caller)
return data
}
func (p *PeerCache) LaunchPeerExecution(peerID string, isMySelf bool, dataID string, url string,
dt utils.DataType, method tools.METHOD, body map[string]interface{}, caller *tools.HTTPCaller) (*PeerExecution, error) {
var err error
b := []byte{}
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 = p.urlFormat(url+meth, dt)
if !p.checkPeerStatus(peerID, caller) {
return nil, errors.New("peer is not reachable")
}
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 {
pexec := &PeerExecution{
Method: method,
Url: url + methods[method],
Body: body,
IsMySelf: isMySelf,
Caller: *caller,
PeerID: peerID,
DataType: dt,
DataID: dataID,
}
singleton.Executions = append(singleton.Executions, pexec)
if currentRountine == 0 {
currentRountine++
go p.retryPeerExecution()
}
return pexec, err
}
fmt.Println("LaunchPeerExecution AFT 3", m, url)
if e, ok := m["error"]; !ok && e != "" {
return nil, errors.New(fmt.Sprintf("%v", m["error"]))
}
return nil, err
}
func (p *PeerCache) retryPeerExecution() {
execs := []*PeerExecution{}
for _, v := range singleton.Executions {
d, err := p.LaunchPeerExecution(v.PeerID, v.IsMySelf, v.DataID, v.Url, v.DataType, v.Method, v.Body, &v.Caller)
if err == nil {
execs = append(execs, d)
} else {
logs.GetLogger().With().Err(err)
}
}
singleton.Executions = execs
if len(singleton.Executions) > 0 {
time.Sleep(60 * time.Second)
p.retryPeerExecution()
} else {
currentRountine = 0
}
}