2 // Copyright 2019 AT&T Intellectual Property
3 // Copyright 2019 Nokia
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
9 // http://www.apache.org/licenses/LICENSE-2.0
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.
17 // This source code is part of the near-RT RIC (RAN Intelligent Controller)
18 // platform project (RICP).
23 // #cgo LDFLAGS: -L/usr/local/lib -lrmr_nng -lnng
24 // #include <rmr/rmr.h>
25 // #include <stdlib.h>
29 "github.com/pkg/errors"
37 func (*Context) Init(readyIntervalSec int, port string, maxMsgSize int, flags int, logger *logger.Logger) RmrMessenger {
39 defer C.free(unsafe.Pointer(pp))
40 logger.Debugf("#rmr_c_go_api.Init - Going to initiate RMR router")
41 ctx := NewContext(maxMsgSize, flags, C.rmr_init(pp, C.int(maxMsgSize), C.int(flags)), logger)
43 ticker := time.NewTicker(time.Duration(readyIntervalSec) * time.Second)
44 for ; !ctx.IsReady(); <-ticker.C {
45 if time.Since(start) >= time.Minute {
46 logger.Debugf("#rmr_c_go_api.Init - Routing table is not ready")
50 logger.Infof("#rmr_c_go_api.Init - RMR router has been initiated")
52 // Configure the rmr to make rounds of attempts to send a message before notifying the application that it should retry.
53 // Each round is about 1000 attempts with a short sleep between each round.
54 C.rmr_set_stimeout(ctx.RmrCtx, C.int(1000))
58 func (ctx *Context) SendMsg(msg *MBuf) (*MBuf, error) {
59 ctx.checkContextInitialized()
60 ctx.Logger.Debugf("#rmr_c_go_api.SendMsg - Going to send message. MBuf: %v", *msg)
61 allocatedCMBuf := ctx.getAllocatedCRmrMBuf(ctx.Logger, msg, ctx.MaxMsgSize)
62 defer C.rmr_free_msg(allocatedCMBuf)
63 state := allocatedCMBuf.state
65 errorMessage := fmt.Sprintf("#rmr_c_go_api.SendMsg - Failed to get allocated message. state: %v - %s", state, states[int(state)])
66 return nil, errors.New(errorMessage)
69 transactionId := string(*msg.XAction)
70 tmpTid := strings.TrimSpace(transactionId)
71 ctx.Logger.Infof("[RSM -> RMR] #rmr_c_go_api.SendMsg - Going to send message %v for transaction id: %s", *msg, tmpTid)
73 currCMBuf := C.rmr_send_msg(ctx.RmrCtx, allocatedCMBuf)
74 state = currCMBuf.state
76 if ctx.Logger.DebugEnabled() {
77 ctx.Logger.Debugf("#rmr_c_go_api.SendMsg - The current message state: %v, message buffer:%v", state, currCMBuf)
81 errorMessage := fmt.Sprintf("#rmr_c_go_api.SendMsg - Failed to send message. state: %v - %s", state, states[int(state)])
82 return nil, errors.New(errorMessage)
85 ctx.Logger.Debugf("#rmr_c_go_api.SendMsg - The message has been sent successfully ")
86 return convertToMBuf(currCMBuf), nil
89 func (ctx *Context) RecvMsg() (*MBuf, error) {
90 ctx.checkContextInitialized()
91 ctx.Logger.Debugf("#rmr_c_go_api.RecvMsg - Going to receive message")
92 allocatedCMBuf := C.rmr_alloc_msg(ctx.RmrCtx, C.int(ctx.MaxMsgSize))
93 defer C.rmr_free_msg(allocatedCMBuf)
95 currCMBuf := C.rmr_rcv_msg(ctx.RmrCtx, allocatedCMBuf)
96 state := currCMBuf.state
99 errorMessage := fmt.Sprintf("#rmr_c_go_api.RecvMsg - Failed to receive message. state: %v - %s", state, states[int(state)])
100 ctx.Logger.Errorf(errorMessage)
101 return nil, errors.New(errorMessage)
104 mbuf := convertToMBuf(currCMBuf)
105 transactionId := string(*mbuf.XAction)
106 tmpTid := strings.TrimSpace(transactionId)
107 ctx.Logger.Infof("[RMR ->RSM] #rmr_c_go_api.RecvMsg - message %v has been received for transaction id: %s", *mbuf, tmpTid)
111 func (ctx *Context) IsReady() bool {
112 ctx.Logger.Debugf("#rmr_c_go_api.IsReady - Going to check if routing table is initialized")
113 return int(C.rmr_ready(ctx.RmrCtx)) != 0
116 func (ctx *Context) Close() {
117 ctx.Logger.Debugf("#rmr_c_go_api.Close - Going to close RMR context")
118 C.rmr_close(ctx.RmrCtx)
119 time.Sleep(100 * time.Millisecond)