Merge "Test case init for e2 reset."
[ric-plt/e2mgr.git] / 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
80         state = currCMBuf.state
81
82         if state != RMR_OK {
83                 errorMessage := fmt.Sprintf("#rmrCgoApi.SendMsg - Failed to send message. state: %v - %s", state, states[int(state)])
84                 return nil, errors.New(errorMessage)
85         }
86
87         return convertToMBuf(ctx.Logger, currCMBuf), nil
88 }
89
90 func (ctx *Context) WhSendMsg(msg *MBuf, printLogs bool) (*MBuf, error) {
91         ctx.checkContextInitialized()
92         ctx.Logger.Debugf("#rmrCgoApi.WhSendMsg - Going to wormhole send message. MBuf: %v", *msg)
93
94         whid := C.rmr_wh_open(ctx.RmrCtx, (*C.char)(msg.GetMsgSrc()))           // open direct connection, returns wormhole ID
95         ctx.Logger.Infof("#rmrCgoApi.WhSendMsg - The wormhole id %v has been received", whid)
96         defer C.rmr_wh_close(ctx.RmrCtx, whid)
97
98         allocatedCMBuf := ctx.getAllocatedCRmrMBuf(ctx.Logger, msg, ctx.MaxMsgSize)
99         state := allocatedCMBuf.state
100         if state != RMR_OK {
101                 errorMessage := fmt.Sprintf("#rmrCgoApi.WhSendMsg - Failed to get allocated message. state: %v - %s", state, states[int(state)])
102                 return nil, errors.New(errorMessage)
103         }
104
105         if printLogs {
106                 transactionId := string(*msg.XAction)
107                 tmpTid := strings.TrimSpace(transactionId)
108                 ctx.Logger.Infof("[E2 Manager -> RMR] #rmrCgoApi.WhSendMsg - Going to send message %v for transaction id: %s", *msg, tmpTid)
109         }
110
111         currCMBuf := C.rmr_wh_send_msg(ctx.RmrCtx, whid, allocatedCMBuf)
112         defer C.rmr_free_msg(currCMBuf)
113
114         state = currCMBuf.state
115
116         if state != RMR_OK {
117                 errorMessage := fmt.Sprintf("#rmrCgoApi.WhSendMsg - Failed to send message. state: %v - %s", state, states[int(state)])
118                 return nil, errors.New(errorMessage)
119         }
120
121         return convertToMBuf(ctx.Logger, currCMBuf), nil
122 }
123
124
125 func (ctx *Context) RecvMsg() (*MBuf, error) {
126         ctx.checkContextInitialized()
127         ctx.Logger.Debugf("#rmrCgoApi.RecvMsg - Going to receive message")
128         allocatedCMBuf := C.rmr_alloc_msg(ctx.RmrCtx, C.int(ctx.MaxMsgSize))
129
130         currCMBuf := C.rmr_rcv_msg(ctx.RmrCtx, allocatedCMBuf)
131         defer C.rmr_free_msg(currCMBuf)
132
133         state := currCMBuf.state
134
135         if state != RMR_OK {
136                 errorMessage := fmt.Sprintf("#rmrCgoApi.RecvMsg - Failed to receive message. state: %v - %s", state, states[int(state)])
137                 ctx.Logger.Errorf(errorMessage)
138                 return nil, errors.New(errorMessage)
139         }
140
141         mbuf := convertToMBuf(ctx.Logger, currCMBuf)
142
143         if mbuf.MType != E2_TERM_KEEP_ALIVE_RESP {
144
145                 transactionId := string(*mbuf.XAction)
146                 tmpTid := strings.TrimSpace(transactionId)
147                 ctx.Logger.Infof("[RMR -> E2 Manager] #rmrCgoApi.RecvMsg - message %v has been received for transaction id: %s", *mbuf, tmpTid)
148         }
149         return mbuf, nil
150 }
151
152 func (ctx *Context) IsReady() bool {
153         ctx.Logger.Debugf("#rmrCgoApi.IsReady - Going to check if routing table is initialized")
154         return int(C.rmr_ready(ctx.RmrCtx)) != 0
155 }
156
157 func (ctx *Context) Close() {
158         ctx.Logger.Debugf("#rmrCgoApi.Close - Going to close RMR context")
159         C.rmr_close(ctx.RmrCtx)
160         time.Sleep(100 * time.Millisecond)
161 }