X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=E2Manager%2FrmrCgo%2FrmrCgoApi.go;h=fc5c849c5d2c215c2d1a7ebf207a02290dd49346;hb=0249b5fc410b6c6814906b185dfdf25b27621148;hp=9b242c42b256ff589789eee6e67d9d45e3bc504a;hpb=549944ecfb9e75ed4a03dd6d10e4160b89f26ca8;p=ric-plt%2Fe2mgr.git diff --git a/E2Manager/rmrCgo/rmrCgoApi.go b/E2Manager/rmrCgo/rmrCgoApi.go index 9b242c4..fc5c849 100644 --- a/E2Manager/rmrCgo/rmrCgoApi.go +++ b/E2Manager/rmrCgo/rmrCgoApi.go @@ -13,11 +13,14 @@ // 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). + package rmrCgo -// #cgo LDFLAGS: -L/usr/local/lib -lrmr_nng -lnng +// #cgo LDFLAGS: -L/usr/local/lib -lrmr_si // #include // #include import "C" @@ -31,7 +34,7 @@ import ( "e2mgr/logger" ) -func (*Context) Init(port string, maxMsgSize int, flags int, logger *logger.Logger) *RmrMessenger {//TODO remove pointer from interface +func (*Context) Init(port string, maxMsgSize int, flags int, logger *logger.Logger) RmrMessenger { pp := C.CString(port) defer C.free(unsafe.Pointer(pp)) logger.Debugf("#rmrCgoApi.Init - Going to initiate RMR router") @@ -46,46 +49,91 @@ func (*Context) Init(port string, maxMsgSize int, flags int, logger *logger.Logg } } logger.Infof("#rmrCgoApi.Init - RMR router has been initiated") + + // Configure the rmr to make rounds of attempts to send a message before notifying the application that it should retry. + // Each round is about 1000 attempts with a short sleep between each round. + C.rmr_set_stimeout(ctx.RmrCtx, C.int(1000)) r := RmrMessenger(ctx) - return &r + return r } -func (ctx *Context) SendMsg(msg *MBuf, maxMsgSize int) (*MBuf, error) { +func (ctx *Context) SendMsg(msg *MBuf, printLogs bool) (*MBuf, error) { ctx.checkContextInitialized() ctx.Logger.Debugf("#rmrCgoApi.SendMsg - Going to send message. MBuf: %v", *msg) - allocatedCMBuf := ctx.getAllocatedCRmrMBuf(ctx.Logger, msg, maxMsgSize) - defer C.rmr_free_msg(allocatedCMBuf) + allocatedCMBuf := ctx.getAllocatedCRmrMBuf(ctx.Logger, msg, ctx.MaxMsgSize) state := allocatedCMBuf.state if state != RMR_OK { errorMessage := fmt.Sprintf("#rmrCgoApi.SendMsg - Failed to get allocated message. state: %v - %s", state, states[int(state)]) return nil, errors.New(errorMessage) } - //TODO: if debug enabled - transactionId := string(*msg.XAction) - tmpTid := strings.TrimSpace(transactionId) - ctx.Logger.Infof("[E2 Manager -> RMR] #rmrCgoApi.SendMsg - Going to send message %v for transaction id: %s", *msg, tmpTid) + if printLogs { + //TODO: if debug enabled + transactionId := string(*msg.XAction) + tmpTid := strings.TrimSpace(transactionId) + ctx.Logger.Infof("[E2 Manager -> RMR] #rmrCgoApi.SendMsg - Going to send message %v for transaction id: %s", *msg, tmpTid) + } currCMBuf := C.rmr_send_msg(ctx.RmrCtx, allocatedCMBuf) + defer C.rmr_free_msg(currCMBuf) + if currCMBuf == nil { + errorMessage := fmt.Sprintf("#rmrCgoApi.SendMsg - currCMBuf is empty hence return from message") + return nil, errors.New(errorMessage) + } + state = currCMBuf.state - ctx.Logger.Debugf("#rmrCgoApi.SendMsg - The current message state: %v, message buffer:%v", state, currCMBuf) if state != RMR_OK { errorMessage := fmt.Sprintf("#rmrCgoApi.SendMsg - Failed to send message. state: %v - %s", state, states[int(state)]) return nil, errors.New(errorMessage) } - ctx.Logger.Debugf("#rmrCgoApi.SendMsg - The message has been sent successfully ") return convertToMBuf(ctx.Logger, currCMBuf), nil } +func (ctx *Context) WhSendMsg(msg *MBuf, printLogs bool) (*MBuf, error) { + ctx.checkContextInitialized() + ctx.Logger.Debugf("#rmrCgoApi.WhSendMsg - Going to wormhole send message. MBuf: %v", *msg) + + whid := C.rmr_wh_open(ctx.RmrCtx, (*C.char)(msg.GetMsgSrc())) // open direct connection, returns wormhole ID + ctx.Logger.Infof("#rmrCgoApi.WhSendMsg - The wormhole id %v has been received", whid) + defer C.rmr_wh_close(ctx.RmrCtx, whid) + + allocatedCMBuf := ctx.getAllocatedCRmrMBuf(ctx.Logger, msg, ctx.MaxMsgSize) + state := allocatedCMBuf.state + if state != RMR_OK { + errorMessage := fmt.Sprintf("#rmrCgoApi.WhSendMsg - Failed to get allocated message. state: %v - %s", state, states[int(state)]) + return nil, errors.New(errorMessage) + } + + if printLogs { + transactionId := string(*msg.XAction) + tmpTid := strings.TrimSpace(transactionId) + ctx.Logger.Infof("[E2 Manager -> RMR] #rmrCgoApi.WhSendMsg - Going to send message %v for transaction id: %s", *msg, tmpTid) + } + + currCMBuf := C.rmr_wh_send_msg(ctx.RmrCtx, whid, allocatedCMBuf) + defer C.rmr_free_msg(currCMBuf) + + state = currCMBuf.state + + if state != RMR_OK { + errorMessage := fmt.Sprintf("#rmrCgoApi.WhSendMsg - Failed to send message. state: %v - %s", state, states[int(state)]) + return nil, errors.New(errorMessage) + } + + return convertToMBuf(ctx.Logger, currCMBuf), nil +} + + func (ctx *Context) RecvMsg() (*MBuf, error) { ctx.checkContextInitialized() ctx.Logger.Debugf("#rmrCgoApi.RecvMsg - Going to receive message") allocatedCMBuf := C.rmr_alloc_msg(ctx.RmrCtx, C.int(ctx.MaxMsgSize)) - defer C.rmr_free_msg(allocatedCMBuf) currCMBuf := C.rmr_rcv_msg(ctx.RmrCtx, allocatedCMBuf) + defer C.rmr_free_msg(currCMBuf) + state := currCMBuf.state if state != RMR_OK { @@ -95,18 +143,14 @@ func (ctx *Context) RecvMsg() (*MBuf, error) { } mbuf := convertToMBuf(ctx.Logger, currCMBuf) - transactionId := string(*mbuf.XAction) - tmpTid := strings.TrimSpace(transactionId) - ctx.Logger.Infof("[RMR -> E2 Manager] #rmrCgoApi.RecvMsg - message %v has been received for transaction id: %s", *mbuf, tmpTid) - return mbuf, nil -} -func (ctx *Context) RtsMsg(msg *MBuf) { - ctx.checkContextInitialized() - ctx.Logger.Debugf("#rmrCgoApi.RtsMsg - Going to return message to the sender") - allocatedCMBuf := C.rmr_alloc_msg(ctx.RmrCtx, C.int(ctx.MaxMsgSize)) - defer C.rmr_free_msg(allocatedCMBuf) - C.rmr_rts_msg(ctx.RmrCtx, allocatedCMBuf) + if mbuf.MType != E2_TERM_KEEP_ALIVE_RESP { + + transactionId := string(*mbuf.XAction) + tmpTid := strings.TrimSpace(transactionId) + ctx.Logger.Infof("[RMR -> E2 Manager] #rmrCgoApi.RecvMsg - message %v has been received for transaction id: %s", *mbuf, tmpTid) + } + return mbuf, nil } func (ctx *Context) IsReady() bool { @@ -118,4 +162,4 @@ func (ctx *Context) Close() { ctx.Logger.Debugf("#rmrCgoApi.Close - Going to close RMR context") C.rmr_close(ctx.RmrCtx) time.Sleep(100 * time.Millisecond) -} +} \ No newline at end of file