Routing manager needs to support for xApps that terminate other protocols than E2

[ric-plt/rtmgr.git] / pkg / sbi / nngpush.go

diff --git a/pkg/sbi/nngpush.go b/pkg/sbi/nngpush.go
index 1f062d3..1f0e0e6 100644 (file)
--- a/pkg/sbi/nngpush.go
+++ b/pkg/sbi/nngpush.go
@@ -14,134 +14,194 @@
    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    See the License for the specific language governing permissions and
    limitations under the License.
    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    See the License for the specific language governing permissions and
    limitations under the License.

+
+   This source code is part of the near-RT RIC (RAN Intelligent Controller)
+   platform project (RICP).
+

 ==================================================================================
 */
 /*
 ==================================================================================
 */
 /*

-  Mnemonic:    nngpipe.go
-  Abstract: mangos (NNG) Pipeline SBI implementation
+  Mnemonic:    rmrpipe.go
+  Abstract: mangos (RMR) Pipeline SBI implementation

   Date:                12 March 2019
 */
 
 package sbi
 
   Date:                12 March 2019
 */
 
 package sbi
 

+/*
+#include <rmr/rmr.h>
+*/
+import "C"
+

 import (
 import (

+       "bytes"
+       "crypto/md5"

        "errors"
        "errors"

-       "nanomsg.org/go/mangos/v2"
-       "nanomsg.org/go/mangos/v2/protocol/push"
-       _ "nanomsg.org/go/mangos/v2/transport/all"
+       "gerrit.o-ran-sc.org/r/ric-plt/xapp-frame/pkg/xapp"

        "routing-manager/pkg/rtmgr"
        "strconv"
        "routing-manager/pkg/rtmgr"
        "strconv"

+       "strings"
+       "fmt"

 )
 
 )
 

-type NngPush struct {
+var rmrcallid = 1
+var rmrdynamiccallid = 201
+
+type RmrPush struct {

        Sbi
        Sbi

-       NewSocket CreateNewNngSocketHandler
+       rcChan chan *xapp.RMRParams

 }
 
 }
 

-func NewNngPush() *NngPush {
-       instance := new(NngPush)
-       instance.NewSocket = createNewPushSocket
-       return instance
+type EPStatus struct {
+        endpoint string
+        status   bool

 }
 
 }
 

-func createNewPushSocket() (NngSocket, error) {
-       rtmgr.Logger.Debug("Invoked: createNewPushSocket()")
-       socket, err := push.NewSocket()
-       if err != nil {
-               return nil, errors.New("can't create new push socket due to:" + err.Error())
-       }
-       socket.SetPipeEventHook(pipeEventHandler)
-       return socket, nil
+type RMRParams struct {
+       *xapp.RMRParams

 }
 
 }
 

-func pipeEventHandler(event mangos.PipeEvent, pipe mangos.Pipe) {
-       rtmgr.Logger.Debug("Invoked: pipeEventHandler()")
-       for _, ep := range rtmgr.Eps {
-               uri := DefaultNngPipelineSocketPrefix + ep.Ip + ":" + strconv.Itoa(DefaultNngPipelineSocketNumber)
-               if uri == pipe.Address() {
-                       switch event {
-                       case 1:
-                               ep.IsReady = true
-                               rtmgr.Logger.Debug("Endpoint " + uri + " successfully attached")
-                       default:
-                               ep.IsReady = false
-                               rtmgr.Logger.Debug("Endpoint " + uri + " has been detached")
-                       }
-               }
-       }
+func (params *RMRParams) String() string {
+       var b bytes.Buffer
+       sum := md5.Sum(params.Payload)
+       fmt.Fprintf(&b, "params(Src=%s Mtype=%d SubId=%d Xid=%s Meid=%s Paylens=%d/%d Payhash=%x)", params.Src, params.Mtype, params.SubId, params.Xid, params.Meid.RanName, params.PayloadLen, len(params.Payload), sum)
+       return b.String()

 }
 
 }
 

-func (c *NngPush) Initialize(ip string) error {
+func NewRmrPush() *RmrPush {
+       instance := new(RmrPush)
+       return instance
+}
+
+func (c *RmrPush) Initialize(ip string) error {

        return nil
 }
 
        return nil
 }
 

-func (c *NngPush) Terminate() error {
+func (c *RmrPush) Terminate() error {

        return nil
 }
 
        return nil
 }
 

-func (c *NngPush) AddEndpoint(ep *rtmgr.Endpoint) error {
-       var err error
-       var socket NngSocket
-       rtmgr.Logger.Debug("Invoked sbi.AddEndpoint")
-       rtmgr.Logger.Debug("args: %v", *ep)
-       socket, err = c.NewSocket()
-       if err != nil {
-               return errors.New("can't add new socket to endpoint:" + ep.Uuid + " due to: " + err.Error())
-       }
-       ep.Socket = socket
-       err = c.dial(ep)
-       if err != nil {
-               return errors.New("can't dial to endpoint:" + ep.Uuid + " due to: " + err.Error())
+func (c *RmrPush) AddEndpoint(ep *rtmgr.Endpoint) error {
+
+       xapp.Logger.Debug("Invoked sbi.AddEndpoint")
+       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))
+       } else {
+               xapp.Logger.Debug("Wormhole ID is %v and EP is %v", ep.Whid, endpoint)

        }
        }

+

        return nil
 }
 
        return nil
 }
 

-func (c *NngPush) DeleteEndpoint(ep *rtmgr.Endpoint) error {
-       rtmgr.Logger.Debug("Invoked sbi. DeleteEndpoint")
-       rtmgr.Logger.Debug("args: %v", *ep)
-       if err := ep.Socket.(NngSocket).Close(); err != nil {
-               return errors.New("can't close push socket of endpoint:" + ep.Uuid + " due to: " + err.Error())
-       }
+func (c *RmrPush) DeleteEndpoint(ep *rtmgr.Endpoint) error {
+       xapp.Logger.Debug("Invoked sbi. DeleteEndpoint")
+       xapp.Logger.Debug("args: %v", *ep)
+
+       xapp.Rmr.Closewh(ep.Whid)

        return nil
 }
 
        return nil
 }
 

-func (c *NngPush) UpdateEndpoints(rcs *rtmgr.RicComponents) {
+func (c *RmrPush) UpdateEndpoints(rcs *rtmgr.RicComponents) {

        c.updateEndpoints(rcs, c)
 }
 
        c.updateEndpoints(rcs, c)
 }
 

-/*
-NOTE: Asynchronous dial starts a goroutine which keep maintains the connection to the given endpoint
-*/
-func (c *NngPush) dial(ep *rtmgr.Endpoint) error {
-       rtmgr.Logger.Debug("Dialing to endpoint: " + ep.Uuid)
-       uri := DefaultNngPipelineSocketPrefix + ep.Ip + ":" + strconv.Itoa(DefaultNngPipelineSocketNumber)
-       options := make(map[string]interface{})
-       options[mangos.OptionDialAsynch] = true
-       if err := ep.Socket.(NngSocket).DialOptions(uri, options); err != nil {
-               return errors.New("can't dial on push socket to " + uri + " due to: " + err.Error())
+func (c *RmrPush) DistributeAll(policies *[]string) error {
+       xapp.Logger.Debug("Invoked: sbi.DistributeAll")
+       xapp.Logger.Debug("args: %v", *policies)
+
+       /*for _, ep := range rtmgr.Eps {
+               go c.send(ep, policies)
+       }*/
+       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)
+                }
+        }
+
+        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
 }
 
        return nil
 }
 

-func (c *NngPush) DistributeAll(policies *[]string) error {
-       rtmgr.Logger.Debug("Invoked: sbi.DistributeAll")
-       rtmgr.Logger.Debug("args: %v", *policies)
-       for _, ep := range rtmgr.Eps {
-               if ep.IsReady {
-                       go c.send(ep, policies)
-               } else {
-                       rtmgr.Logger.Warn("Endpoint " + ep.Uuid + " is not ready")
-               }
-       }
-       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)
+
+       ret := c.send_data(ep, policies, call_id)
+
+        channel <- EPStatus{ep.Uuid, ret}
+

 }
 
 }
 

-func (c *NngPush) send(ep *rtmgr.Endpoint, policies *[]string) {
-       rtmgr.Logger.Debug("Push policy to endpoint: " + ep.Uuid)
-       for _, pe := range *policies {
-               if err := ep.Socket.(NngSocket).Send([]byte(pe)); err != nil {
-                       rtmgr.Logger.Error("Unable to send policy entry due to: " + err.Error())
-               }
+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
+
+        var policy = []byte{}
+
+        for _, pe := range *policies {
+                b:= []byte(pe)
+                for j:=0; j<len(b); j++{
+                        policy = append(policy,b[j])
+                }

        }
        }

-       rtmgr.Logger.Info("NNG PUSH to endpoint " + ep.Uuid + ": OK (# of Entries:" + strconv.Itoa(len(*policies)) + ")")
+        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
+        }
+
+        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) CreateEndpoint(payload string) *rtmgr.Endpoint {
+       return c.createEndpoint(payload, c)
+}
+
+func (c *RmrPush) DistributeToEp(policies *[]string, ep *rtmgr.Endpoint) error {
+       xapp.Logger.Debug("Invoked: sbi.DistributeToEp")
+       xapp.Logger.Debug("args: %v", *policies)
+
+       if rmrdynamiccallid == 255 {
+                rmrdynamiccallid = 201
+        }
+
+       go c.send_data(ep, policies,rmrdynamiccallid)
+       rmrdynamiccallid++
+
+       return nil

 }
 }