X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=pkg%2Fsbi%2Fnngpush.go;h=0d9634eab06c3848389507dc7e7f0ac0412a2a9c;hb=765b273b4444673d4e45a5f363dea28cdb6ca0bf;hp=1f0e0e6fa7d1ce6db934f37ad2d9a15555c03782;hpb=2c131acf0418c45d9045a6a81baf3e7d047fcb38;p=ric-plt%2Frtmgr.git

diff --git a/pkg/sbi/nngpush.go b/pkg/sbi/nngpush.go
index 1f0e0e6..0d9634e 100644
--- a/pkg/sbi/nngpush.go
+++ b/pkg/sbi/nngpush.go
@@ -37,15 +37,20 @@ import (
 	"bytes"
 	"crypto/md5"
 	"errors"
+	"fmt"
 	"gerrit.o-ran-sc.org/r/ric-plt/xapp-frame/pkg/xapp"
 	"routing-manager/pkg/rtmgr"
 	"strconv"
 	"strings"
-	"fmt"
+	"sync"
+	"time"
 )
 
 var rmrcallid = 1
 var rmrdynamiccallid = 201
+var addendpointct = 1
+var count int
+var Conn sync.Mutex
 
 type RmrPush struct {
 	Sbi
@@ -53,8 +58,8 @@ type RmrPush struct {
 }
 
 type EPStatus struct {
-        endpoint string
-        status   bool
+	endpoint string
+	status   bool
 }
 
 type RMRParams struct {
@@ -82,12 +87,17 @@ func (c *RmrPush) Terminate() error {
 }
 
 func (c *RmrPush) AddEndpoint(ep *rtmgr.Endpoint) error {
-
-	xapp.Logger.Debug("Invoked sbi.AddEndpoint")
+	addendpointct = addendpointct + 1
+	count := addendpointct
+	xapp.Logger.Debug("Invoked sbi.AddEndpoint for %s with count = %d", ep.Ip, count)
 	endpoint := ep.Ip + ":" + strconv.Itoa(DefaultRmrPipelineSocketNumber)
 	ep.Whid = int(xapp.Rmr.Openwh(endpoint))
 	if ep.Whid < 0 {
-		return errors.New("can't open warmhole connection for endpoint:" + ep.Uuid + " due to invalid Wormhole ID: " + string(ep.Whid))
+		time.Sleep(time.Duration(10) * time.Second)
+		ep.Whid = int(xapp.Rmr.Openwh(endpoint))
+		if ep.Whid < 0 {
+			return errors.New("cannot open wormhole connection for endpoint:" + ep.Uuid + " due to invalid Wormhole ID: " + string(ep.Whid) + " count: " + strconv.Itoa(count))
+		}
 	} else {
 		xapp.Logger.Debug("Wormhole ID is %v and EP is %v", ep.Whid, endpoint)
 	}
@@ -114,94 +124,142 @@ func (c *RmrPush) DistributeAll(policies *[]string) error {
 	/*for _, ep := range rtmgr.Eps {
 		go c.send(ep, policies)
 	}*/
-	channel := make(chan EPStatus)
+	//channel := make(chan EPStatus)
 
 	if rmrcallid == 200 {
 		rmrcallid = 1
 	}
 
-        for _, ep := range rtmgr.Eps {
-                go c.send_sync(ep,  policies, channel, rmrcallid)
-        }
-	rmrcallid++
-
-        count := 0
-        result := make([]EPStatus, len(rtmgr.Eps))
-        for i, _ := range result {
-                result[i] = <-channel
-                if result[i].status == true {
-                        count++
-                } else {
-                        xapp.Logger.Error("RMR send failed for endpoint %v", result[i].endpoint)
-                }
-        }
+	for _, ep := range rtmgr.Eps {
+		go c.send_sync(ep, policies, rmrcallid)
+	}
 
-        if count < len(rtmgr.Eps) {
-                return errors.New(" RMR response count " + string(count) + " is less than half of endpoint list " + string(len(rtmgr.Eps)))
-        }
+	rmrcallid++
 
+	/*
+				count := 0
+		        result := make([]EPStatus, len(rtmgr.Eps))
+		        for i, _ := range result {
+		                result[i] = <-channel
+		                if result[i].status == true {
+		                        count++
+		                } else {
+		                        xapp.Logger.Error("RMR sent failed for endpoint %v", result[i].endpoint)
+		                }
+		        }
+
+		        if count < len(rtmgr.Eps) {
+		                return errors.New(" RMR response count " + string(count) + " is less than half of endpoint list " + string(len(rtmgr.Eps)))
+		        }*/
 
 	return nil
 }
 
-func (c *RmrPush) send_sync(ep *rtmgr.Endpoint, policies *[]string, channel chan EPStatus, call_id int) {
-        xapp.Logger.Debug("Push policy to endpoint: " + ep.Uuid)
+//func (c *RmrPush) send_sync(ep *rtmgr.Endpoint, policies *[]string, channel chan EPStatus, call_id int) {
+func (c *RmrPush) send_sync(ep *rtmgr.Endpoint, policies *[]string, call_id int) {
+	xapp.Logger.Debug("Push policy to endpoint: " + ep.Uuid)
 
 	ret := c.send_data(ep, policies, call_id)
+	for count = 0; count <= 2 && ret == false; count++ {
+		time.Sleep(time.Second)
+		ret := c.send_data(ep, policies, call_id)
+		if ret == true {
+			break
+		}
+		xapp.Logger.Error("Invoked  send_data to try again due to return value : %v", ret)
+	}
 
-        channel <- EPStatus{ep.Uuid, ret}
+	Conn.Lock()
+	rtmgr.RMRConnStatus[ep.Uuid] = ret
+	Conn.Unlock()
+	// Handling per connection .. may be updating global map
+
+	//channel <- EPStatus{ep.Uuid, ret}
 
 }
 
 func (c *RmrPush) send_data(ep *rtmgr.Endpoint, policies *[]string, call_id int) bool {
-        xapp.Logger.Debug("Invoked send_data to endpoint: " + ep.Uuid + " call_id: " + strconv.Itoa(call_id))
-        var state int
-        var retstr string
+	xapp.Logger.Debug("Invoked send_data to endpoint: " + ep.Uuid + " call_id: " + strconv.Itoa(call_id))
+	var state int
+	var retstr string
 
-        var policy = []byte{}
+	var policy = []byte{}
 
-        for _, pe := range *policies {
-                b:= []byte(pe)
-                for j:=0; j<len(b); j++{
-                        policy = append(policy,b[j])
-                }
+	for _, pe := range *policies {
+		b := []byte(pe)
+		for j := 0; j < len(b); j++ {
+			policy = append(policy, b[j])
+		}
 	}
-        params := &RMRParams{&xapp.RMRParams{}}
-        params.Mtype = 20
-        params.PayloadLen = len(policy)
-        params.Payload =[]byte(policy)
-        params.Mbuf = nil
-        params.Whid = ep.Whid
-        params.Callid = call_id
-        params.Timeout = 200
-        state, retstr = xapp.Rmr.SendCallMsg(params.RMRParams)
-	routestatus := strings.Split(retstr," ")
-        if state != C.RMR_OK && routestatus[0] == "OK" {
-              xapp.Logger.Error("Updating Routes to Endpoint: " + ep.Uuid + " failed, call_id: " + strconv.Itoa(call_id) + " for xapp.Rmr.SendCallMsg " + " Route Update Status: " + routestatus[0])
-              return false
-        } else {
+	params := &RMRParams{&xapp.RMRParams{}}
+	params.Mtype = 20
+	params.PayloadLen = len(policy)
+	params.Payload = []byte(policy)
+	params.Mbuf = nil
+	params.Whid = ep.Whid
+	params.Callid = call_id
+	params.Timeout = 200
+	state, retstr = xapp.Rmr.SendCallMsg(params.RMRParams)
+	routestatus := strings.Split(retstr, " ")
+	if state != C.RMR_OK && routestatus[0] != "OK" {
+		xapp.Logger.Error("Updating Routes to Endpoint: " + ep.Uuid + " failed, call_id: " + strconv.Itoa(call_id) + " for xapp.Rmr.SendCallMsg " + " Route Update Status: " + routestatus[0])
+		return false
+	} else {
 		xapp.Logger.Info("Update Routes to Endpoint: " + ep.Uuid + " successful, call_id: " + strconv.Itoa(call_id) + ", Payload length: " + strconv.Itoa(params.PayloadLen) + ", Route Update Status: " + routestatus[0] + "(# of Entries:" + strconv.Itoa(len(*policies)))
-              return true
-        }
+		return true
+	}
+}
 
-        xapp.Logger.Error("Route Update to endpoint: " + ep.Uuid + " failed, call_id: " + strconv.Itoa(call_id) + " xapp.Rmr.SendCallMsg not called")
-        return false
+func (c *RmrPush) CheckEndpoint(payload string) (ep *rtmgr.Endpoint) {
+	return c.checkEndpoint(payload)
 }
 
-func (c *RmrPush) CreateEndpoint(payload string) *rtmgr.Endpoint {
-	return c.createEndpoint(payload, c)
+func (c *RmrPush) CreateEndpoint(rmrsrc string) (ep *string, whid int) {
+	return c.createEndpoint(rmrsrc)
 }
 
-func (c *RmrPush) DistributeToEp(policies *[]string, ep *rtmgr.Endpoint) error {
+func (c *RmrPush) DistributeToEp(policies *[]string, ep string, whid int) error {
 	xapp.Logger.Debug("Invoked: sbi.DistributeToEp")
 	xapp.Logger.Debug("args: %v", *policies)
 
 	if rmrdynamiccallid == 255 {
-                rmrdynamiccallid = 201
-        }
+		rmrdynamiccallid = 201
+	}
 
-	go c.send_data(ep, policies,rmrdynamiccallid)
+	go c.sendDynamicRoutes(ep, whid, policies, rmrdynamiccallid)
 	rmrdynamiccallid++
 
 	return nil
 }
+
+func (c *RmrPush) sendDynamicRoutes(ep string, whid int, policies *[]string, call_id int) bool {
+	xapp.Logger.Debug("Invoked send_rt_process to endpoint: " + ep + " call_id: " + strconv.Itoa(call_id) + "whid: " + strconv.Itoa(whid))
+	var state int
+	var retstr string
+
+	var policy = []byte{}
+
+	for _, pe := range *policies {
+		b := []byte(pe)
+		for j := 0; j < len(b); j++ {
+			policy = append(policy, b[j])
+		}
+	}
+	params := &RMRParams{&xapp.RMRParams{}}
+	params.Mtype = 20
+	params.PayloadLen = len(policy)
+	params.Payload = []byte(policy)
+	params.Mbuf = nil
+	params.Whid = whid
+	params.Callid = call_id
+	params.Timeout = 200
+	state, retstr = xapp.Rmr.SendCallMsg(params.RMRParams)
+	routestatus := strings.Split(retstr, " ")
+	if state != C.RMR_OK && routestatus[0] != "OK" {
+		xapp.Logger.Error("Updating Routes to Endpoint: " + ep + " failed, call_id: " + strconv.Itoa(call_id) + ",whi_id: " + strconv.Itoa(whid) + " for xapp.Rmr.SendCallMsg " + " Route Update Status: " + routestatus[0])
+		return false
+	} else {
+		xapp.Logger.Info("Update Routes to Endpoint: " + ep + " successful, call_id: " + strconv.Itoa(call_id) + ", Payload length: " + strconv.Itoa(params.PayloadLen) + ",whid: " + strconv.Itoa(whid) + ", Route Update Status: " + routestatus[0] + "(# of Entries:" + strconv.Itoa(len(*policies)))
+		return true
+	}
+}