oc-lib/models/peer/peer_cache.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
	}
}