E2Manager/rmrCgo/rmrCgoApi.go

   1 //
   2 // Copyright 2019 AT&T Intellectual Property
   3 // Copyright 2019 Nokia
   4 //
   5 // Licensed under the Apache License, Version 2.0 (the "License");
   6 // you may not use this file except in compliance with the License.
   7 // You may obtain a copy of the License at
   8 //
   9 //      http://www.apache.org/licenses/LICENSE-2.0
  10 //
  11 // Unless required by applicable law or agreed to in writing, software
  12 // distributed under the License is distributed on an "AS IS" BASIS,
  13 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14 // See the License for the specific language governing permissions and
  15 // limitations under the License.
  16
  17 //  This source code is part of the near-RT RIC (RAN Intelligent Controller)
  18 //  platform project (RICP).
  19
  20
  21 package rmrCgo
  22
  23 // #cgo LDFLAGS: -L/usr/local/lib -lrmr_si
  24 // #include <rmr/rmr.h>
  25 // #include <stdlib.h>
  26 import "C"
  27 import (
  28         "fmt"
  29         "github.com/pkg/errors"
  30         "strings"
  31         "time"
  32         "unsafe"
  33
  34         "e2mgr/logger"
  35 )
  36
  37 func (*Context) Init(port string, maxMsgSize int, flags int, logger *logger.Logger) RmrMessenger {
  38         pp := C.CString(port)
  39         defer C.free(unsafe.Pointer(pp))
  40         logger.Debugf("#rmrCgoApi.Init - Going to initiate RMR router")
  41         ctx := NewContext(maxMsgSize, flags, C.rmr_init(pp, C.int(maxMsgSize), C.int(flags)), logger)
  42         start := time.Now()
  43         //TODO use time.Ticker()
  44         for !ctx.IsReady() {
  45                 time.Sleep(time.Second)
  46                 if time.Since(start) >= time.Minute {
  47                         logger.Debugf("#rmrCgoApi.Init - Routing table is not ready")
  48                         start = time.Now()
  49                 }
  50         }
  51         logger.Infof("#rmrCgoApi.Init - RMR router has been initiated")
  52
  53         // Configure the rmr to make rounds of attempts to send a message before notifying the application that it should retry.
  54         // Each round is about 1000 attempts with a short sleep between each round.
  55         C.rmr_set_stimeout(ctx.RmrCtx, C.int(1000))
  56         r := RmrMessenger(ctx)
  57         return r
  58 }
  59
  60 func (ctx *Context) SendMsg(msg *MBuf, printLogs bool) (*MBuf, error) {
  61         ctx.checkContextInitialized()
  62         ctx.Logger.Debugf("#rmrCgoApi.SendMsg - Going to send message. MBuf: %v", *msg)
  63         allocatedCMBuf := ctx.getAllocatedCRmrMBuf(ctx.Logger, msg, ctx.MaxMsgSize)
  64         state := allocatedCMBuf.state
  65         if state != RMR_OK {
  66                 errorMessage := fmt.Sprintf("#rmrCgoApi.SendMsg - Failed to get allocated message. state: %v - %s", state, states[int(state)])
  67                 return nil, errors.New(errorMessage)
  68         }
  69
  70         if printLogs {
  71                 //TODO: if debug enabled
  72                 transactionId := string(*msg.XAction)
  73                 tmpTid := strings.TrimSpace(transactionId)
  74                 ctx.Logger.Infof("[E2 Manager -> RMR] #rmrCgoApi.SendMsg - Going to send message %v for transaction id: %s", *msg, tmpTid)
  75         }
  76
  77         currCMBuf := C.rmr_send_msg(ctx.RmrCtx, allocatedCMBuf)
  78         defer C.rmr_free_msg(currCMBuf)
  79         if currCMBuf == nil {
  80                 errorMessage := fmt.Sprintf("#rmrCgoApi.SendMsg - currCMBuf is empty hence return from message")
  81                 return nil, errors.New(errorMessage)
  82         }
  83
  84         state = currCMBuf.state
  85
  86         if state != RMR_OK {
  87                 errorMessage := fmt.Sprintf("#rmrCgoApi.SendMsg - Failed to send message. state: %v - %s", state, states[int(state)])
  88                 return nil, errors.New(errorMessage)
  89         }
  90
  91         return convertToMBuf(ctx.Logger, currCMBuf), nil
  92 }
  93
  94 func (ctx *Context) WhSendMsg(msg *MBuf, printLogs bool) (*MBuf, error) {
  95         ctx.checkContextInitialized()
  96         ctx.Logger.Debugf("#rmrCgoApi.WhSendMsg - Going to wormhole send message. MBuf: %v", *msg)
  97
  98         whid := C.rmr_wh_open(ctx.RmrCtx, (*C.char)(msg.GetMsgSrc()))           // open direct connection, returns wormhole ID
  99         ctx.Logger.Infof("#rmrCgoApi.WhSendMsg - The wormhole id %v has been received", whid)
 100         defer C.rmr_wh_close(ctx.RmrCtx, whid)
 101
 102         allocatedCMBuf := ctx.getAllocatedCRmrMBuf(ctx.Logger, msg, ctx.MaxMsgSize)
 103         state := allocatedCMBuf.state
 104         if state != RMR_OK {
 105                 errorMessage := fmt.Sprintf("#rmrCgoApi.WhSendMsg - Failed to get allocated message. state: %v - %s", state, states[int(state)])
 106                 return nil, errors.New(errorMessage)
 107         }
 108
 109         if printLogs {
 110                 transactionId := string(*msg.XAction)
 111                 tmpTid := strings.TrimSpace(transactionId)
 112                 ctx.Logger.Infof("[E2 Manager -> RMR] #rmrCgoApi.WhSendMsg - Going to send message %v for transaction id: %s", *msg, tmpTid)
 113         }
 114
 115         currCMBuf := C.rmr_wh_send_msg(ctx.RmrCtx, whid, allocatedCMBuf)
 116         defer C.rmr_free_msg(currCMBuf)
 117
 118         state = currCMBuf.state
 119
 120         if state != RMR_OK {
 121                 errorMessage := fmt.Sprintf("#rmrCgoApi.WhSendMsg - Failed to send message. state: %v - %s", state, states[int(state)])
 122                 return nil, errors.New(errorMessage)
 123         }
 124
 125         return convertToMBuf(ctx.Logger, currCMBuf), nil
 126 }
 127
 128
 129 func (ctx *Context) RecvMsg() (*MBuf, error) {
 130         ctx.checkContextInitialized()
 131         ctx.Logger.Debugf("#rmrCgoApi.RecvMsg - Going to receive message")
 132         allocatedCMBuf := C.rmr_alloc_msg(ctx.RmrCtx, C.int(ctx.MaxMsgSize))
 133
 134         currCMBuf := C.rmr_rcv_msg(ctx.RmrCtx, allocatedCMBuf)
 135         defer C.rmr_free_msg(currCMBuf)
 136
 137         state := currCMBuf.state
 138
 139         if state != RMR_OK {
 140                 errorMessage := fmt.Sprintf("#rmrCgoApi.RecvMsg - Failed to receive message. state: %v - %s", state, states[int(state)])
 141                 ctx.Logger.Errorf(errorMessage)
 142                 return nil, errors.New(errorMessage)
 143         }
 144
 145         mbuf := convertToMBuf(ctx.Logger, currCMBuf)
 146
 147         if mbuf.MType != E2_TERM_KEEP_ALIVE_RESP {
 148
 149                 transactionId := string(*mbuf.XAction)
 150                 tmpTid := strings.TrimSpace(transactionId)
 151                 ctx.Logger.Infof("[RMR -> E2 Manager] #rmrCgoApi.RecvMsg - message %v has been received for transaction id: %s", *mbuf, tmpTid)
 152         }
 153         return mbuf, nil
 154 }
 155
 156 func (ctx *Context) IsReady() bool {
 157         ctx.Logger.Debugf("#rmrCgoApi.IsReady - Going to check if routing table is initialized")
 158         return int(C.rmr_ready(ctx.RmrCtx)) != 0
 159 }
 160
 161 func (ctx *Context) Close() {
 162         ctx.Logger.Debugf("#rmrCgoApi.Close - Going to close RMR context")
 163         C.rmr_close(ctx.RmrCtx)
 164         time.Sleep(100 * time.Millisecond)
 165 }